ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From voze...@apache.org
Subject [08/12] ignite git commit: IGNITE-2206: intermediate saving commit.
Date Wed, 23 Dec 2015 10:40:50 GMT
IGNITE-2206: intermediate saving commit.


Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/89690022
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/89690022
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/89690022

Branch: refs/heads/ignite-2206
Commit: 89690022863a32fb00e76335c02c6a36bfae55e1
Parents: ba11b3a
Author: iveselovskiy <iveselovskiy@gridgain.com>
Authored: Tue Dec 22 22:51:07 2015 +0300
Committer: iveselovskiy <iveselovskiy@gridgain.com>
Committed: Tue Dec 22 22:51:07 2015 +0300

----------------------------------------------------------------------
 .../processors/hadoop/HadoopPayloadAware.java   |  12 ++
 .../processors/hadoop/PayloadAware.java         |  13 --
 .../internal/processors/igfs/IgfsImpl.java      |  12 +-
 .../internal/processors/igfs/IgfsPaths.java     |  37 +---
 .../hadoop/fs/HadoopFileSystemFactory.java      |  29 ---
 .../fs/IgniteHadoopIgfsSecondaryFileSystem.java |  19 +-
 .../fs/v1/DefaultHadoopFileSystemFactory.java   | 185 ++++++++++++++++++
 .../hadoop/fs/v1/HadoopFileSystemFactory.java   |  21 +++
 .../hadoop/fs/v1/IgniteHadoopFileSystem.java    |  27 ++-
 .../fs/v2/HadoopAbstractFileSystemFactory.java  |  21 +++
 .../hadoop/fs/v2/IgniteHadoopFileSystem.java    |   7 +-
 .../fs/DefaultHadoopFileSystemFactory.java      | 187 -------------------
 .../hadoop/fs/HadoopLazyConcurrentMap.java      |   4 +
 .../IgniteHadoopFileSystemAbstractSelfTest.java |  12 +-
 ...teHadoopFileSystemShmemAbstractSelfTest.java |   2 -
 15 files changed, 284 insertions(+), 304 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/ignite/blob/89690022/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopPayloadAware.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopPayloadAware.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopPayloadAware.java
new file mode 100644
index 0000000..dcb163f
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopPayloadAware.java
@@ -0,0 +1,12 @@
+package org.apache.ignite.internal.processors.hadoop;
+
+/**
+ *
+ */
+public interface HadoopPayloadAware {
+    /**
+     *
+     * @return
+     */
+    public Object getPayload();
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/89690022/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/PayloadAware.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/PayloadAware.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/PayloadAware.java
deleted file mode 100644
index dc4ff5e..0000000
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/PayloadAware.java
+++ /dev/null
@@ -1,13 +0,0 @@
-package org.apache.ignite.internal.processors.hadoop;
-
-/**
- * Created by ivan on 22.12.15.
- */
-public interface PayloadAware <P> {
-
-    /**
-     *
-     * @return
-     */
-    public P getPayload();
-}

http://git-wip-us.apache.org/repos/asf/ignite/blob/89690022/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsImpl.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsImpl.java
index fb93ea1..7453e15 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsImpl.java
@@ -19,7 +19,6 @@ package org.apache.ignite.internal.processors.igfs;
 
 import java.io.IOException;
 import java.io.OutputStream;
-import java.io.Serializable;
 import java.net.URI;
 import java.util.ArrayList;
 import java.util.Collection;
@@ -73,7 +72,7 @@ import org.apache.ignite.internal.IgniteKernal;
 import org.apache.ignite.internal.managers.communication.GridMessageListener;
 import org.apache.ignite.internal.managers.eventstorage.GridEventStorageManager;
 import org.apache.ignite.internal.managers.eventstorage.GridLocalEventListener;
-import org.apache.ignite.internal.processors.hadoop.PayloadAware;
+import org.apache.ignite.internal.processors.hadoop.HadoopPayloadAware;
 import org.apache.ignite.internal.processors.task.GridInternal;
 import org.apache.ignite.internal.util.GridSpinBusyLock;
 import org.apache.ignite.internal.util.future.GridCompoundFuture;
@@ -125,7 +124,7 @@ public final class IgfsImpl implements IgfsEx {
     static final Map<String, String> DFLT_DIR_META = F.asMap(PROP_PERMISSION, PERMISSION_DFLT_VAL);
 
     /** Handshake message. */
-    private final IgfsPaths<Serializable> secondaryPaths;
+    private final IgfsPaths secondaryPaths;
 
     /** Cache based structure (meta data) manager. */
     private IgfsMetaManager meta;
@@ -260,11 +259,10 @@ public final class IgfsImpl implements IgfsEx {
 
         modeRslvr = new IgfsModeResolver(dfltMode, modes);
 
-        Serializable secondaryFsPayload = null;
+        Object secondaryFsPayload = null;
 
-        if (secondaryFs instanceof PayloadAware) {
-            secondaryFsPayload = ((PayloadAware<Serializable>) secondaryFs).getPayload();
-        }
+        if (secondaryFs instanceof HadoopPayloadAware)
+            secondaryFsPayload = ((HadoopPayloadAware) secondaryFs).getPayload();
 
         secondaryPaths = new IgfsPaths(
             //secondaryFs == null ? null : secondaryFs.properties(),

http://git-wip-us.apache.org/repos/asf/ignite/blob/89690022/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsPaths.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsPaths.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsPaths.java
index d434d01..83451db 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsPaths.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsPaths.java
@@ -25,12 +25,10 @@ import java.io.ObjectInput;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutput;
 import java.io.ObjectOutputStream;
-import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.List;
 import org.apache.ignite.igfs.IgfsMode;
 import org.apache.ignite.igfs.IgfsPath;
-import org.apache.ignite.internal.processors.hadoop.PayloadAware;
 import org.apache.ignite.internal.util.typedef.T2;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.jetbrains.annotations.Nullable;
@@ -38,16 +36,12 @@ import org.jetbrains.annotations.Nullable;
 /**
  * Description of path modes.
  */
-public class IgfsPaths <P extends Serializable> implements Externalizable, PayloadAware<P>
{
+public class IgfsPaths implements Externalizable {
     /** */
     private static final long serialVersionUID = 0L;
 
-//    /** Additional secondary file system properties. */
-//    @Deprecated
-//    private Map<String, String> props;
-
     /** */
-    private P payload;
+    private Object payload;
 
     /** Default IGFS mode. */
     private IgfsMode dfltMode;
@@ -68,26 +62,14 @@ public class IgfsPaths <P extends Serializable> implements Externalizable,
Paylo
      * @param dfltMode Default IGFS mode.
      * @param pathModes Path modes.
      */
-    public IgfsPaths(//Map<String, String> props,
-                     P payload,
+    public IgfsPaths(Object payload,
                      IgfsMode dfltMode,
                      @Nullable List<T2<IgfsPath, IgfsMode>> pathModes) {
-        //this.props = props;
         this.payload = payload;
         this.dfltMode = dfltMode;
         this.pathModes = pathModes;
     }
 
-//    /**
-//     * @return Secondary file system properties.
-//     *
-//     * @deprecated
-//     */
-//    @Deprecated
-//    public Map<String, String> properties() {
-//        return props;
-//    }
-
     /**
      * @return Default IGFS mode.
      */
@@ -102,15 +84,6 @@ public class IgfsPaths <P extends Serializable> implements Externalizable,
Paylo
         return pathModes;
     }
 
-//    /**
-//     * Getter for factory.
-//     *
-//     * @return The factory.
-//     */
-//    public HadoopFileSystemFactory<F> factory() {
-//        return factory;
-//    }
-
     /** {@inheritDoc} */
     @Override public void writeExternal(ObjectOutput out) throws IOException {
 //        U.writeStringMap(out, props);
@@ -188,14 +161,14 @@ public class IgfsPaths <P extends Serializable> implements Externalizable,
Paylo
         ObjectInput oi = new ObjectInputStream(new ByteArrayInputStream(factoryBytes));
 
         try {
-            payload = (P) oi.readObject();
+            payload = oi.readObject();
         }
         finally {
             oi.close();
         }
     }
 
-    @Override public P getPayload() {
+    public Object getPayload() {
         return payload;
     }
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ignite/blob/89690022/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/HadoopFileSystemFactory.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/HadoopFileSystemFactory.java
b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/HadoopFileSystemFactory.java
deleted file mode 100644
index 5337f12..0000000
--- a/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/HadoopFileSystemFactory.java
+++ /dev/null
@@ -1,29 +0,0 @@
-package org.apache.ignite.hadoop.fs;
-
-import java.io.IOException;
-import java.io.Serializable;
-
-/**
- * This factory is {@link Serializable} because it should be transferable over the network.
- *
- * @param <T> The type
- */
-public interface HadoopFileSystemFactory <T> extends Serializable {
-    /**
-     * Gets the file system, possibly creating it or taking a cached instance.
-     * All the other data needed for the file system creation are expected to be contained
-     * in this object instance.
-     *
-     * @param userName The user name
-     * @return The file system.
-     * @throws IOException On error.
-     */
-    public T get(String userName) throws IOException;
-
-//    /**
-//     * Getter for the file system URI.
-//     *
-//     * @return The file system URI.
-//     */
-//    public URI uri();
-}

http://git-wip-us.apache.org/repos/asf/ignite/blob/89690022/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/IgniteHadoopIgfsSecondaryFileSystem.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/IgniteHadoopIgfsSecondaryFileSystem.java
b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/IgniteHadoopIgfsSecondaryFileSystem.java
index 7ba136b..86ed7a0 100644
--- a/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/IgniteHadoopIgfsSecondaryFileSystem.java
+++ b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/IgniteHadoopIgfsSecondaryFileSystem.java
@@ -35,6 +35,7 @@ import org.apache.hadoop.fs.PathIsNotEmptyDirectoryException;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteException;
+import org.apache.ignite.hadoop.fs.v1.HadoopFileSystemFactory;
 import org.apache.ignite.igfs.IgfsDirectoryNotEmptyException;
 import org.apache.ignite.igfs.IgfsException;
 import org.apache.ignite.igfs.IgfsFile;
@@ -45,8 +46,8 @@ import org.apache.ignite.igfs.IgfsPathNotFoundException;
 import org.apache.ignite.igfs.IgfsUserContext;
 import org.apache.ignite.igfs.secondary.IgfsSecondaryFileSystem;
 import org.apache.ignite.igfs.secondary.IgfsSecondaryFileSystemPositionedReadable;
-import org.apache.ignite.internal.processors.hadoop.PayloadAware;
-import org.apache.ignite.internal.processors.hadoop.fs.DefaultHadoopFileSystemFactory;
+import org.apache.ignite.internal.processors.hadoop.HadoopPayloadAware;
+import org.apache.ignite.hadoop.fs.v1.DefaultHadoopFileSystemFactory;
 import org.apache.ignite.internal.processors.hadoop.igfs.HadoopIgfsProperties;
 import org.apache.ignite.internal.processors.hadoop.igfs.HadoopIgfsSecondaryFileSystemPositionedReadable;
 import org.apache.ignite.internal.processors.igfs.IgfsFileImpl;
@@ -69,7 +70,7 @@ import static org.apache.ignite.internal.processors.igfs.IgfsEx.PROP_USER_NAME;
  * see {@link IgfsUserContext#currentUser()}.
  */
 public class IgniteHadoopIgfsSecondaryFileSystem implements IgfsSecondaryFileSystem,
-        LifecycleAware, PayloadAware<HadoopFileSystemFactory<FileSystem>> {
+        LifecycleAware, HadoopPayloadAware {
 //    /** Properties of file system, see {@link #properties()}
 //     * */
 //    private final Map<String, String> props = new HashMap<>();
@@ -97,7 +98,7 @@ public class IgniteHadoopIgfsSecondaryFileSystem implements IgfsSecondaryFileSys
     private String dfltUserName = IgfsUtils.fixUserName(null);
 
     /** */
-    private HadoopFileSystemFactory<FileSystem> fsFactory;
+    private HadoopFileSystemFactory fsFactory;
 
     private final AtomicBoolean started = new AtomicBoolean();
 
@@ -174,20 +175,18 @@ public class IgniteHadoopIgfsSecondaryFileSystem implements IgfsSecondaryFileSys
             fac.setUri(uri);
 
         if (cfgPath != null)
-            fac.setCfgPaths(Collections.singletonList(cfgPath));
+            fac.setConfigPaths(Collections.singletonList(cfgPath));
 
         setFsFactory(fac);
 
         setDfltUserName(userName);
-
-        start();
     }
 
     /**
      *
      * @param factory
      */
-    public void setFsFactory(HadoopFileSystemFactory<FileSystem> factory) {
+    public void setFsFactory(HadoopFileSystemFactory factory) {
         A.ensure(factory != null, "Factory value must not be null.");
 
         this.fsFactory = factory;
@@ -600,7 +599,6 @@ public class IgniteHadoopIgfsSecondaryFileSystem implements IgfsSecondaryFileSys
 
     /**
      * Should be invoked by client (from Spring?) after all the setters invoked.
-     * TODO: how this should be invoked?
      *
      * @throws IgniteCheckedException
      */
@@ -609,6 +607,7 @@ public class IgniteHadoopIgfsSecondaryFileSystem implements IgfsSecondaryFileSys
         A.ensure(fsFactory != null, "factory");
         A.ensure(dfltUserName != null, "dfltUserName");
 
+        // Avoid
         if (started.compareAndSet(false, true)) {
             if (fsFactory instanceof LifecycleAware)
                 ((LifecycleAware) fsFactory).start();
@@ -638,7 +637,7 @@ public class IgniteHadoopIgfsSecondaryFileSystem implements IgfsSecondaryFileSys
         close();
     }
 
-    @Override public HadoopFileSystemFactory<FileSystem> getPayload() {
+    @Override public HadoopFileSystemFactory getPayload() {
         return fsFactory;
     }
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ignite/blob/89690022/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v1/DefaultHadoopFileSystemFactory.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v1/DefaultHadoopFileSystemFactory.java
b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v1/DefaultHadoopFileSystemFactory.java
new file mode 100644
index 0000000..ba4bfdd
--- /dev/null
+++ b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v1/DefaultHadoopFileSystemFactory.java
@@ -0,0 +1,185 @@
+package org.apache.ignite.hadoop.fs.v1;
+
+import java.io.Externalizable;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.internal.processors.hadoop.HadoopUtils;
+import org.apache.ignite.internal.processors.hadoop.fs.HadoopFileSystemsUtils;
+import org.apache.ignite.internal.processors.hadoop.fs.HadoopLazyConcurrentMap;
+import org.apache.ignite.internal.processors.igfs.IgfsPaths;
+import org.apache.ignite.internal.processors.igfs.IgfsUtils;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.A;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lifecycle.LifecycleAware;
+
+import static org.apache.ignite.internal.util.lang.GridFunc.nullifyEmpty;
+
+/**
+ * The class is to be instantiated as a Spring beans, so it must have public zero-arg constructor.
+ * The class is serializable as it will be transferred over the network as a part of {@link
IgfsPaths} object.
+ */
+public class DefaultHadoopFileSystemFactory implements HadoopFileSystemFactory, Externalizable,
LifecycleAware {
+    /** Lazy per-user cache for the file systems. It is cleared and nulled in #close() method.
*/
+    private final transient HadoopLazyConcurrentMap<String, FileSystem> fileSysLazyMap
= new HadoopLazyConcurrentMap<>(
+        new HadoopLazyConcurrentMap.ValueFactory<String, FileSystem>() {
+            @Override public FileSystem createValue(String key) {
+                try {
+                    assert !F.isEmpty(key);
+
+                    return createFileSystem(key);
+                }
+                catch (IOException ioe) {
+                    throw new IgniteException(ioe);
+                }
+            }
+        }
+    );
+
+    /** Configuration of the secondary filesystem, never null. */
+    protected transient Configuration cfg;
+
+    /** */
+    protected transient URI uri;
+
+    /** */
+    protected String uriStr;
+
+    /** */
+    protected List<String> cfgPathStr;
+
+    int getCount = 0;
+
+    /**
+     *
+     */
+    public DefaultHadoopFileSystemFactory() {
+        //
+
+
+
+    }
+
+    @Override public FileSystem get(String userName) throws IOException {
+        A.ensure(cfg != null, "cfg");
+
+        if (getCount == 0)
+            assert fileSysLazyMap.size() == 0;
+
+        getCount++;
+
+        return fileSysLazyMap.getOrCreate(userName);
+    }
+
+    /**
+     * Uri setter.
+     * @param uriStr
+     */
+    public void setUri(String uriStr) {
+        this.uriStr = uriStr;
+    }
+
+    /**
+     * Configuration(s) setter, to be invoked from Spring config.
+     * @param cfgPaths
+     */
+    public void setConfigPaths(List<String> cfgPaths) {
+        this.cfgPathStr = (List)nullifyEmpty(cfgPaths);
+    }
+
+    /**
+     * @return {@link org.apache.hadoop.fs.FileSystem}  instance for this secondary Fs.
+     * @throws IOException
+     */
+    protected FileSystem createFileSystem(String userName) throws IOException {
+        userName = IgfsUtils.fixUserName(nullifyEmpty(userName));
+
+        assert cfg != null;
+
+        final FileSystem fileSys;
+
+        try {
+            fileSys = FileSystem.get(uri, cfg, userName);
+        }
+        catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+
+            throw new IOException("Failed to create file system due to interrupt.", e);
+        }
+
+        return fileSys;
+    }
+
+    @Override public void writeExternal(ObjectOutput out) throws IOException {
+        U.writeString(out, uriStr);
+
+        U.writeCollection(out, cfgPathStr);
+    }
+
+    @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
+        uriStr = U.readString(in);
+
+        cfgPathStr = new ArrayList(U.readCollection(in));
+    }
+
+    @Override public void start() throws IgniteException {
+        cfg = HadoopUtils.safeCreateConfiguration();
+
+        if (cfgPathStr != null) {
+            for (String confPath : cfgPathStr) {
+                confPath = nullifyEmpty(confPath);
+
+                if (confPath != null) {
+                    URL url = U.resolveIgniteUrl(confPath);
+
+                    if (url == null) {
+                        // If secConfPath is given, it should be resolvable:
+                        throw new IllegalArgumentException("Failed to resolve secondary file
system configuration path " +
+
+                            "(ensure that it exists locally and you have read access to it):
" + confPath);
+                    }
+
+                    cfg.addResource(url);
+                }
+            }
+        }
+
+        // if secondary fs URI is not given explicitly, try to get it from the configuration:
+        if (uriStr == null)
+            uri = FileSystem.getDefaultUri(cfg);
+        else {
+            try {
+                uri = new URI(uriStr);
+            }
+            catch (URISyntaxException use) {
+                throw new IgniteException("Failed to resolve secondary file system URI: "
+ uriStr);
+            }
+        }
+
+        assert uriStr != null;
+
+        // Disable caching:
+        String prop = HadoopFileSystemsUtils.disableFsCachePropertyName(uri.getScheme());
+
+        cfg.setBoolean(prop, true);
+    }
+
+    @Override public void stop() throws IgniteException {
+        try {
+            fileSysLazyMap.close();
+        }
+        catch (IgniteCheckedException ice) {
+            throw new IgniteException(ice);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/89690022/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v1/HadoopFileSystemFactory.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v1/HadoopFileSystemFactory.java
b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v1/HadoopFileSystemFactory.java
new file mode 100644
index 0000000..c1c7b9d
--- /dev/null
+++ b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v1/HadoopFileSystemFactory.java
@@ -0,0 +1,21 @@
+package org.apache.ignite.hadoop.fs.v1;
+
+import java.io.IOException;
+import java.io.Serializable;
+import org.apache.hadoop.fs.FileSystem;
+
+/**
+ * This factory is {@link Serializable} because it should be transferable over the network.
+ */
+public interface HadoopFileSystemFactory extends Serializable {
+    /**
+     * Gets the file system, possibly creating it or taking a cached instance.
+     * All the other data needed for the file system creation are expected to be contained
+     * in this object instance.
+     *
+     * @param userName The user name
+     * @return The file system.
+     * @throws IOException On error.
+     */
+    public FileSystem get(String userName) throws IOException;
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/89690022/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v1/IgniteHadoopFileSystem.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v1/IgniteHadoopFileSystem.java
b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v1/IgniteHadoopFileSystem.java
index 932e326..355892e 100644
--- a/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v1/IgniteHadoopFileSystem.java
+++ b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v1/IgniteHadoopFileSystem.java
@@ -18,12 +18,9 @@
 package org.apache.ignite.hadoop.fs.v1;
 
 import java.io.BufferedOutputStream;
-import java.io.ByteArrayInputStream;
 import java.io.File;
 import java.io.FileNotFoundException;
 import java.io.IOException;
-import java.io.ObjectInput;
-import java.io.ObjectInputStream;
 import java.io.OutputStream;
 import java.net.URI;
 import java.net.URISyntaxException;
@@ -47,7 +44,6 @@ import org.apache.hadoop.hdfs.DFSUtil;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.util.Progressable;
 import org.apache.ignite.IgniteException;
-import org.apache.ignite.hadoop.fs.HadoopFileSystemFactory;
 import org.apache.ignite.igfs.IgfsBlockLocation;
 import org.apache.ignite.igfs.IgfsException;
 import org.apache.ignite.igfs.IgfsFile;
@@ -71,6 +67,7 @@ import org.apache.ignite.internal.util.typedef.X;
 import org.apache.ignite.internal.util.typedef.internal.A;
 import org.apache.ignite.internal.util.typedef.internal.S;
 import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lifecycle.LifecycleAware;
 import org.jetbrains.annotations.Nullable;
 
 import static org.apache.ignite.configuration.FileSystemConfiguration.DFLT_IGFS_LOG_BATCH_SIZE;
@@ -190,7 +187,8 @@ public class IgniteHadoopFileSystem extends FileSystem {
     /** {@inheritDoc} */
     @Override public URI getUri() {
         if (uri == null)
-            throw new IllegalStateException("URI is null (was IgniteHadoopFileSystem properly
initialized?).");
+            throw new IllegalStateException("URI is null (was IgniteHadoopFileSystem properly
initialized?) [closed="
+                + closeGuard.get() + ']');
 
         return uri;
     }
@@ -243,6 +241,8 @@ public class IgniteHadoopFileSystem extends FileSystem {
     @Override public void initialize(URI name, Configuration cfg) throws IOException {
         enterBusy();
 
+        assert !closeGuard.get();
+
         try {
             if (rmtClient != null)
                 throw new IOException("File system is already initialized: " + rmtClient);
@@ -295,7 +295,7 @@ public class IgniteHadoopFileSystem extends FileSystem {
 
             igfsGrpBlockSize = handshake.blockSize();
 
-            final IgfsPaths<HadoopFileSystemFactory<FileSystem>> paths = handshake.secondaryPaths();
+            final IgfsPaths paths = handshake.secondaryPaths();
 
             // Initialize client logger.
             Boolean logEnabled = parameter(cfg, PARAM_IGFS_LOG_ENABLED, uriAuthority, false);
@@ -336,16 +336,15 @@ public class IgniteHadoopFileSystem extends FileSystem {
 
 //                byte[] secFsFacoryBytes = handshake.getSecondaryFileSystemFactoryBytes();
 
-                HadoopFileSystemFactory<FileSystem> factory = paths.getPayload();
+                HadoopFileSystemFactory factory = (HadoopFileSystemFactory)paths.getPayload();
 
                 A.ensure(factory != null, "Secondary file system factory should not be null.");
 
-                //secondaryUri = factory.uri();
+                if (factory instanceof LifecycleAware)
+                    ((LifecycleAware) factory).start();
 
                 try {
-                    //SecondaryFileSystemProvider secProvider = new SecondaryFileSystemProvider(secUri,
secConfPath);
-
-                    secondaryFs = factory.get(user); //secProvider.createFileSystem(user);
+                    secondaryFs = factory.get(user);
 
                     secondaryUri = secondaryFs.getUri();
 
@@ -440,12 +439,12 @@ public class IgniteHadoopFileSystem extends FileSystem {
         if (clientLog.isLogEnabled())
             clientLog.close();
 
-        if (secondaryFs != null)
-            U.closeQuiet(secondaryFs);
+        U.closeQuiet(secondaryFs);
+
+        System.out.println("closed " + uri);
 
         // Reset initialized resources.
         uri = null;
-        System.out.println("uri zeroed.");
         rmtClient = null;
     }
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/89690022/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v2/HadoopAbstractFileSystemFactory.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v2/HadoopAbstractFileSystemFactory.java
b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v2/HadoopAbstractFileSystemFactory.java
new file mode 100644
index 0000000..cf81e57
--- /dev/null
+++ b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v2/HadoopAbstractFileSystemFactory.java
@@ -0,0 +1,21 @@
+package org.apache.ignite.hadoop.fs.v2;
+
+import java.io.IOException;
+import java.io.Serializable;
+import org.apache.hadoop.fs.AbstractFileSystem;
+
+/**
+ * This factory is {@link Serializable} because it should be transferable over the network.
+ */
+interface HadoopAbstractFileSystemFactory extends Serializable {
+    /**
+     * Gets the file system, possibly creating it or taking a cached instance.
+     * All the other data needed for the file system creation are expected to be contained
+     * in this object instance.
+     *
+     * @param userName The user name
+     * @return The file system.
+     * @throws IOException On error.
+     */
+    public AbstractFileSystem get(String userName) throws IOException;
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/89690022/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v2/IgniteHadoopFileSystem.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v2/IgniteHadoopFileSystem.java
b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v2/IgniteHadoopFileSystem.java
index d3267c7..96f97dc 100644
--- a/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v2/IgniteHadoopFileSystem.java
+++ b/modules/hadoop/src/main/java/org/apache/ignite/hadoop/fs/v2/IgniteHadoopFileSystem.java
@@ -52,7 +52,6 @@ import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.util.DataChecksum;
 import org.apache.hadoop.util.Progressable;
 import org.apache.ignite.IgniteException;
-import org.apache.ignite.hadoop.fs.HadoopFileSystemFactory;
 import org.apache.ignite.igfs.IgfsBlockLocation;
 import org.apache.ignite.igfs.IgfsFile;
 import org.apache.ignite.igfs.IgfsMode;
@@ -300,7 +299,7 @@ public class IgniteHadoopFileSystem extends AbstractFileSystem implements
Closea
 
             grpBlockSize = handshake.blockSize();
 
-            IgfsPaths<HadoopFileSystemFactory<AbstractFileSystem>> paths = handshake.secondaryPaths();
+            IgfsPaths paths = handshake.secondaryPaths();
 
             Boolean logEnabled = parameter(cfg, PARAM_IGFS_LOG_ENABLED, uriAuthority, false);
 
@@ -338,8 +337,8 @@ public class IgniteHadoopFileSystem extends AbstractFileSystem implements
Closea
 //                String secUri = props.get(SECONDARY_FS_URI);
 //                String secConfPath = props.get(SECONDARY_FS_CONFIG_PATH);
 
-                HadoopFileSystemFactory<AbstractFileSystem> factory
-                    = (HadoopFileSystemFactory<AbstractFileSystem>)paths.getPayload();
+                HadoopAbstractFileSystemFactory factory
+                    = (HadoopAbstractFileSystemFactory)paths.getPayload();
 
                 A.ensure(secondaryUri != null, "File system factory uri should not be null.");
 

http://git-wip-us.apache.org/repos/asf/ignite/blob/89690022/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/fs/DefaultHadoopFileSystemFactory.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/fs/DefaultHadoopFileSystemFactory.java
b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/fs/DefaultHadoopFileSystemFactory.java
deleted file mode 100644
index bee0f25..0000000
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/fs/DefaultHadoopFileSystemFactory.java
+++ /dev/null
@@ -1,187 +0,0 @@
-package org.apache.ignite.internal.processors.hadoop.fs;
-
-import java.io.Externalizable;
-import java.io.IOException;
-import java.io.ObjectInput;
-import java.io.ObjectOutput;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.net.URL;
-import java.util.Collection;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.IgniteException;
-import org.apache.ignite.hadoop.fs.HadoopFileSystemFactory;
-import org.apache.ignite.internal.processors.hadoop.HadoopUtils;
-import org.apache.ignite.internal.processors.igfs.IgfsPaths;
-import org.apache.ignite.internal.processors.igfs.IgfsUtils;
-import org.apache.ignite.internal.util.typedef.F;
-import org.apache.ignite.internal.util.typedef.internal.A;
-import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.lifecycle.LifecycleAware;
-
-import static org.apache.ignite.internal.util.lang.GridFunc.nullifyEmpty;
-
-/**
- * The class is to be instantiated as a Spring beans, so it must have public zero-arg constructor.
- * The class is serializable as it will be transferred over the network as a part of {@link
IgfsPaths} object.
- */
-public class DefaultHadoopFileSystemFactory implements HadoopFileSystemFactory<FileSystem>,
Externalizable, LifecycleAware {
-    /** Configuration of the secondary filesystem, never null. */
-    protected final Configuration cfg = HadoopUtils.safeCreateConfiguration();
-
-    /** */
-    private URI uri;
-
-    /** Lazy per-user cache for the file systems. It is cleared and nulled in #close() method.
*/
-    private final HadoopLazyConcurrentMap<String, FileSystem> fileSysLazyMap = new
HadoopLazyConcurrentMap<>(
-        new HadoopLazyConcurrentMap.ValueFactory<String, FileSystem>() {
-            @Override public FileSystem createValue(String key) {
-                try {
-                    assert !F.isEmpty(key);
-
-                    return createFileSystem(key);
-                }
-                catch (IOException ioe) {
-                    throw new IgniteException(ioe);
-                }
-            }
-        }
-    );
-
-    public DefaultHadoopFileSystemFactory() {
-        //
-    }
-
-    @Override public FileSystem get(String userName) throws IOException {
-        return fileSysLazyMap.getOrCreate(userName);
-    }
-
-    public void setUri(URI uri) {
-        this.uri = uri;
-    }
-
-    /**
-     * Convenience mathod, analog of {@link #setUri(URI)} with String type argument.
-     * @param uriStr
-     */
-    public void setUri(String uriStr) {
-        try {
-            setUri(new URI(uriStr));
-        }
-        catch (URISyntaxException use) {
-            throw new IgniteException(use);
-        }
-    }
-
-    /**
-     * Configuration(s) setter, to be invoked from Spring config.
-     * @param cfgPaths
-     */
-    public void setCfgPaths(Collection<String> cfgPaths) {
-        cfgPaths = nullifyEmpty(cfgPaths);
-
-        if (cfgPaths == null)
-            return;
-
-        for (String confPath: cfgPaths) {
-            confPath = nullifyEmpty(confPath);
-
-            if (confPath != null) {
-                URL url = U.resolveIgniteUrl(confPath);
-
-                if (url == null) {
-                    // If secConfPath is given, it should be resolvable:
-                    throw new IllegalArgumentException("Failed to resolve secondary file
system configuration path " +
-                        "(ensure that it exists locally and you have read access to it):
" + confPath);
-                }
-
-                cfg.addResource(url);
-            }
-        }
-    }
-
-    protected void init() throws IOException {
-        String secUri = nullifyEmpty(uri == null ? null : uri.toString());
-
-        A.ensure(cfg != null, "config");
-
-        // if secondary fs URI is not given explicitly, try to get it from the configuration:
-        if (secUri == null)
-            uri = FileSystem.getDefaultUri(cfg);
-        else {
-            try {
-                uri = new URI(secUri);
-            }
-            catch (URISyntaxException use) {
-                throw new IOException("Failed to resolve secondary file system URI: " + secUri);
-            }
-        }
-
-        // Disable caching:
-        String prop = HadoopFileSystemsUtils.disableFsCachePropertyName(uri.getScheme());
-
-        cfg.setBoolean(prop, true);
-    }
-
-    /**
-     * @return {@link org.apache.hadoop.fs.FileSystem}  instance for this secondary Fs.
-     * @throws IOException
-     */
-    protected FileSystem createFileSystem(String userName) throws IOException {
-        userName = IgfsUtils.fixUserName(nullifyEmpty(userName));
-
-        final FileSystem fileSys;
-
-        try {
-            fileSys = FileSystem.get(uri, cfg, userName);
-        }
-        catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-
-            throw new IOException("Failed to create file system due to interrupt.", e);
-        }
-
-        return fileSys;
-    }
-
-    @Override public void writeExternal(ObjectOutput out) throws IOException {
-        cfg.write(out);
-
-        U.writeString(out, uri.toString());
-    }
-
-    @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException
{
-        cfg.clear();
-
-        cfg.readFields(in);
-
-        String uriStr = U.readString(in);
-
-        try {
-            uri = new URI(uriStr);
-        }
-        catch (URISyntaxException use) {
-            throw new IOException(use);
-        }
-    }
-
-    @Override public void start() throws IgniteException {
-        try {
-            init();
-        }
-        catch (IOException ice) {
-            throw new IgniteException(ice);
-        }
-    }
-
-    @Override public void stop() throws IgniteException {
-        try {
-            fileSysLazyMap.close();
-        }
-        catch (IgniteCheckedException ice) {
-            throw new IgniteException(ice);
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/ignite/blob/89690022/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/fs/HadoopLazyConcurrentMap.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/fs/HadoopLazyConcurrentMap.java
b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/fs/HadoopLazyConcurrentMap.java
index 89eaf73..58b5120 100644
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/fs/HadoopLazyConcurrentMap.java
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/fs/HadoopLazyConcurrentMap.java
@@ -57,6 +57,10 @@ public class HadoopLazyConcurrentMap<K, V extends Closeable> {
         assert getClass().getClassLoader() == Ignite.class.getClassLoader();
     }
 
+    public int size () {
+        return map.size();
+    }
+
     /**
      * Gets cached or creates a new value of V.
      * Never returns null.

http://git-wip-us.apache.org/repos/asf/ignite/blob/89690022/modules/hadoop/src/test/java/org/apache/ignite/igfs/IgniteHadoopFileSystemAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/igfs/IgniteHadoopFileSystemAbstractSelfTest.java
b/modules/hadoop/src/test/java/org/apache/ignite/igfs/IgniteHadoopFileSystemAbstractSelfTest.java
index 7e5ef39..1ce0492 100644
--- a/modules/hadoop/src/test/java/org/apache/ignite/igfs/IgniteHadoopFileSystemAbstractSelfTest.java
+++ b/modules/hadoop/src/test/java/org/apache/ignite/igfs/IgniteHadoopFileSystemAbstractSelfTest.java
@@ -25,7 +25,6 @@ import java.io.IOException;
 import java.io.OutputStream;
 import java.lang.reflect.Field;
 import java.net.URI;
-import java.net.URISyntaxException;
 import java.security.PrivilegedExceptionAction;
 import java.util.ArrayDeque;
 import java.util.Arrays;
@@ -63,7 +62,7 @@ import org.apache.ignite.configuration.FileSystemConfiguration;
 import org.apache.ignite.configuration.IgniteConfiguration;
 import org.apache.ignite.hadoop.fs.IgniteHadoopIgfsSecondaryFileSystem;
 import org.apache.ignite.hadoop.fs.v1.IgniteHadoopFileSystem;
-import org.apache.ignite.internal.processors.hadoop.fs.DefaultHadoopFileSystemFactory;
+import org.apache.ignite.hadoop.fs.v1.DefaultHadoopFileSystemFactory;
 import org.apache.ignite.internal.processors.hadoop.igfs.HadoopIgfsEx;
 import org.apache.ignite.internal.processors.hadoop.igfs.HadoopIgfsIpcIo;
 import org.apache.ignite.internal.processors.hadoop.igfs.HadoopIgfsOutProc;
@@ -385,16 +384,16 @@ public abstract class IgniteHadoopFileSystemAbstractSelfTest extends
IgfsCommonA
 
         if (mode != PRIMARY) {
             DefaultHadoopFileSystemFactory fac = new DefaultHadoopFileSystemFactory();
+
             fac.setUri(SECONDARY_URI);
-            fac.setCfgPaths(Collections.singletonList(SECONDARY_CFG_PATH));
+            fac.setConfigPaths(Collections.singletonList(SECONDARY_CFG_PATH));
 
             IgniteHadoopIgfsSecondaryFileSystem sec = new IgniteHadoopIgfsSecondaryFileSystem();
 
             sec.setFsFactory(fac);
             sec.setDfltUserName(SECONDARY_FS_USER);
 
-            sec.start();
-
+            // NB: start() will be invoked upon IgfsImpl init.
             cfg.setSecondaryFileSystem(sec);
         }
 
@@ -412,7 +411,8 @@ public abstract class IgniteHadoopFileSystemAbstractSelfTest extends IgfsCommonA
             @Override public Object call() throws Exception {
                 return new IgniteHadoopFileSystem().getUri();
             }
-        }, IllegalStateException.class, "URI is null (was IgniteHadoopFileSystem properly
initialized?).");
+        }, IllegalStateException.class,
+            "URI is null (was IgniteHadoopFileSystem properly initialized?) [closed=false]");
     }
 
     /** @throws Exception If failed. */

http://git-wip-us.apache.org/repos/asf/ignite/blob/89690022/modules/hadoop/src/test/java/org/apache/ignite/igfs/IgniteHadoopFileSystemShmemAbstractSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/igfs/IgniteHadoopFileSystemShmemAbstractSelfTest.java
b/modules/hadoop/src/test/java/org/apache/ignite/igfs/IgniteHadoopFileSystemShmemAbstractSelfTest.java
index 20c2bd2..d8cf74c 100644
--- a/modules/hadoop/src/test/java/org/apache/ignite/igfs/IgniteHadoopFileSystemShmemAbstractSelfTest.java
+++ b/modules/hadoop/src/test/java/org/apache/ignite/igfs/IgniteHadoopFileSystemShmemAbstractSelfTest.java
@@ -60,8 +60,6 @@ public abstract class IgniteHadoopFileSystemShmemAbstractSelfTest extends
Ignite
      */
     @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
     public void testOutOfResources() throws Exception {
-        if (1 == 1) return;
-
         final Collection<IpcEndpoint> eps = new LinkedList<>();
 
         try {


Mime
View raw message