IGNITE-2206: Hadoop file system creation is now abstracted out using …

…factory interface.

modules/core/src/main/java/org/apache/ignite/igfs/IgfsUserContext.java

@@ -34,24 +34,24 @@ public abstract class IgfsUserContext {
      * The main contract of this method is that {@link #currentUser()} method invoked
      * inside closure always returns 'user' this callable executed with.
      * @param user the user name to invoke closure on behalf of.
-     * @param clo the closure to execute
+     * @param c the closure to execute
      * @param <T> The type of closure result.
      * @return the result of closure execution.
      * @throws IllegalArgumentException if user name is null or empty String or if the closure is null.
      */
-    public static <T> T doAs(String user, final IgniteOutClosure<T> clo) {
+    public static <T> T doAs(String user, final IgniteOutClosure<T> c) {
         if (F.isEmpty(user))
             throw new IllegalArgumentException("Failed to use null or empty user name.");
 
         final String ctxUser = userStackThreadLocal.get();
 
         if (F.eq(ctxUser, user))
-            return clo.apply(); // correct context is already there
+            return c.apply(); // correct context is already there
 
         userStackThreadLocal.set(user);
 
         try {
-            return clo.apply();
+            return c.apply();
         }
         finally {
             userStackThreadLocal.set(ctxUser);
@@ -81,24 +81,24 @@ public static <T> T doAs(String user, final IgniteOutClosure<T> clo) {
      *  }
      * </pre>
      * @param user the user name to invoke closure on behalf of.
-     * @param clbl the Callable to execute
+     * @param c the Callable to execute
      * @param <T> The type of callable result.
      * @return the result of closure execution.
      * @throws IllegalArgumentException if user name is null or empty String or if the closure is null.
      */
-    public static <T> T doAs(String user, final Callable<T> clbl) throws Exception {
+    public static <T> T doAs(String user, final Callable<T> c) throws Exception {
         if (F.isEmpty(user))
             throw new IllegalArgumentException("Failed to use null or empty user name.");
 
         final String ctxUser = userStackThreadLocal.get();
 
         if (F.eq(ctxUser, user))
-            return clbl.call(); // correct context is already there
+            return c.call(); // correct context is already there
 
         userStackThreadLocal.set(user);
 
         try {
-            return clbl.call();
+            return c.call();
         }
         finally {
             userStackThreadLocal.set(ctxUser);

modules/core/src/main/java/org/apache/ignite/igfs/secondary/IgfsSecondaryFileSystem.java

@@ -192,18 +192,4 @@ public OutputStream append(IgfsPath path, int bufSize, boolean create, @Nullable
      * @throws IgniteException In case of error.
      */
     public long usedSpaceSize() throws IgniteException;
-
-    /**
-     * Gets the implementation specific properties of file system.
-     *
-     * @return Map of properties.
-     */
-    public Map<String,String> properties();
-
-
-    /**
-     * Closes the secondary file system.
-     * @throws IgniteException in case of an error.
-     */
-    public void close() throws IgniteException;
 }

modules/core/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopPayloadAware.java

@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.hadoop;
+
+/**
+ * Gets payload for Hadoop secondary file system.
+ */
+public interface HadoopPayloadAware {
+    /**
+     * @return Payload.
+     */
+    public Object getPayload();
+}

modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsEx.java

@@ -43,19 +43,6 @@ public interface IgfsEx extends IgniteFileSystem {
     /** File property: prefer writes to local node. */
     public static final String PROP_PREFER_LOCAL_WRITES = "locWrite";
 
-    /** Property name for path to Hadoop configuration. */
-    public static final String SECONDARY_FS_CONFIG_PATH = "SECONDARY_FS_CONFIG_PATH";
-
-    /** Property name for URI of file system. */
-    public static final String SECONDARY_FS_URI = "SECONDARY_FS_URI";
-
-    /** Property name for default user name of file system.
-     * NOTE: for secondary file system this is just a default user name, which is used
-     * when the 2ndary filesystem is used outside of any user context.
-     * If another user name is set in the context, 2ndary file system will work on behalf
-     * of that user, which is different from the default. */
-     public static final String SECONDARY_FS_USER_NAME = "SECONDARY_FS_USER_NAME";
-
     /**
      * Stops IGFS cleaning all used resources.
      *

modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsImpl.java

@@ -72,6 +72,7 @@
 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.HadoopPayloadAware;
 import org.apache.ignite.internal.processors.task.GridInternal;
 import org.apache.ignite.internal.util.GridSpinBusyLock;
 import org.apache.ignite.internal.util.future.GridCompoundFuture;
@@ -87,6 +88,7 @@
 import org.apache.ignite.lang.IgniteBiTuple;
 import org.apache.ignite.lang.IgniteFuture;
 import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.lifecycle.LifecycleAware;
 import org.apache.ignite.resources.IgniteInstanceResource;
 import org.apache.ignite.thread.IgniteThreadPoolExecutor;
 import org.jetbrains.annotations.Nullable;
@@ -200,6 +202,9 @@ public final class IgfsImpl implements IgfsEx {
         data = igfsCtx.data();
         secondaryFs = cfg.getSecondaryFileSystem();
 
+        if (secondaryFs instanceof LifecycleAware)
+            ((LifecycleAware) secondaryFs).start();
+
         /* Default IGFS mode. */
         IgfsMode dfltMode;
 
@@ -256,8 +261,12 @@ public final class IgfsImpl implements IgfsEx {
 
         modeRslvr = new IgfsModeResolver(dfltMode, modes);
 
-        secondaryPaths = new IgfsPaths(secondaryFs == null ? null : secondaryFs.properties(), dfltMode,
-            modeRslvr.modesOrdered());
+        Object secondaryFsPayload = null;
+
+        if (secondaryFs instanceof HadoopPayloadAware)
+            secondaryFsPayload = ((HadoopPayloadAware) secondaryFs).getPayload();
+
+        secondaryPaths = new IgfsPaths(secondaryFsPayload, dfltMode, modeRslvr.modesOrdered());
 
         // Check whether IGFS LRU eviction policy is set on data cache.
         String dataCacheName = igfsCtx.configuration().getDataCacheName();
@@ -305,7 +314,8 @@ private ClusterNode localNode() {
                 batch.cancel();
 
             try {
-                secondaryFs.close();
+                if (secondaryFs instanceof LifecycleAware)
+                    ((LifecycleAware)secondaryFs).stop();
             }
             catch (Exception e) {
                 log.error("Failed to close secondary file system.", e);

modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsPaths.java

@@ -17,17 +17,21 @@
 
 package org.apache.ignite.internal.processors.igfs;
 
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
 import java.io.Externalizable;
 import java.io.IOException;
 import java.io.ObjectInput;
 import java.io.ObjectOutput;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.Map;
+
+import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.igfs.IgfsMode;
 import org.apache.ignite.igfs.IgfsPath;
 import org.apache.ignite.internal.util.typedef.T2;
 import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.marshaller.jdk.JdkMarshaller;
 import org.jetbrains.annotations.Nullable;
 
 /**
@@ -37,8 +41,8 @@ public class IgfsPaths implements Externalizable {
     /** */
     private static final long serialVersionUID = 0L;
 
-    /** Additional secondary file system properties. */
-    private Map<String, String> props;
+    /** */
+    private byte[] payloadBytes;
 
     /** Default IGFS mode. */
     private IgfsMode dfltMode;
@@ -56,22 +60,25 @@ public IgfsPaths() {
     /**
      * Constructor.
      *
-     * @param props Additional secondary file system properties.
+     * @param payload Payload.
      * @param dfltMode Default IGFS mode.
      * @param pathModes Path modes.
+     * @throws IgniteCheckedException If failed.
      */
-    public IgfsPaths(Map<String, String> props, IgfsMode dfltMode, @Nullable List<T2<IgfsPath,
-        IgfsMode>> pathModes) {
-        this.props = props;
+    public IgfsPaths(Object payload, IgfsMode dfltMode, @Nullable List<T2<IgfsPath, IgfsMode>> pathModes)
+        throws IgniteCheckedException {
         this.dfltMode = dfltMode;
         this.pathModes = pathModes;
-    }
 
-    /**
-     * @return Secondary file system properties.
-     */
-    public Map<String, String> properties() {
-        return props;
+        if (payload == null)
+            payloadBytes = null;
+        else {
+            ByteArrayOutputStream out = new ByteArrayOutputStream();
+
+            new JdkMarshaller().marshal(payload, out);
+
+            payloadBytes = out.toByteArray();
+        }
     }
 
     /**
@@ -88,17 +95,36 @@ public IgfsMode defaultMode() {
         return pathModes;
     }
 
+    /**
+     * @return Payload.
+     *
+     * @throws IgniteCheckedException If failed to deserialize the payload.
+     */
+    @Nullable public Object getPayload(ClassLoader clsLdr) throws IgniteCheckedException {
+        if (payloadBytes == null)
+            return null;
+        else {
+            ByteArrayInputStream in = new ByteArrayInputStream(payloadBytes);
+
+            return new JdkMarshaller().unmarshal(in, clsLdr);
+        }
+    }
+
     /** {@inheritDoc} */
     @Override public void writeExternal(ObjectOutput out) throws IOException {
-        U.writeStringMap(out, props);
+        U.writeByteArray(out, payloadBytes);
+
         U.writeEnum(out, dfltMode);
 
         if (pathModes != null) {
             out.writeBoolean(true);
             out.writeInt(pathModes.size());
 
             for (T2<IgfsPath, IgfsMode> pathMode : pathModes) {
+                assert pathMode.getKey() != null;
+
                 pathMode.getKey().writeExternal(out);
+
                 U.writeEnum(out, pathMode.getValue());
             }
         }
@@ -108,7 +134,8 @@ public IgfsMode defaultMode() {
 
     /** {@inheritDoc} */
     @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
-        props = U.readStringMap(in);
+        payloadBytes = U.readByteArray(in);
+
         dfltMode = IgfsMode.fromOrdinal(in.readByte());
 
         if (in.readBoolean()) {
@@ -118,11 +145,10 @@ public IgfsMode defaultMode() {
 
             for (int i = 0; i < size; i++) {
                 IgfsPath path = new IgfsPath();
-                path.readExternal(in);
 
-                T2<IgfsPath, IgfsMode> entry = new T2<>(path, IgfsMode.fromOrdinal(in.readByte()));
+                path.readExternal(in);
 
-                pathModes.add(entry);
+                pathModes.add(new T2<>(path, IgfsMode.fromOrdinal(in.readByte())));
             }
         }
     }

modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsSecondaryFileSystemImpl.java

@@ -19,7 +19,6 @@
 
 import java.io.OutputStream;
 import java.util.Collection;
-import java.util.Collections;
 import java.util.Map;
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.igfs.IgfsFile;
@@ -116,14 +115,4 @@ class IgfsSecondaryFileSystemImpl implements IgfsSecondaryFileSystem {
     @Override public long usedSpaceSize() throws IgniteException {
         return igfs.usedSpaceSize();
     }
-
-    /** {@inheritDoc} */
-    @Override public Map<String, String> properties() {
-        return Collections.emptyMap();
-    }
-
-    /** {@inheritDoc} */
-    @Override public void close() throws IgniteException {
-        // No-op.
-    }
 }