ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From sboi...@apache.org
Subject [42/63] [abbrv] ignite git commit: IGNITE-3912: Hadoop: Implemented new class loading architecture for embedded execution mode.
Date Tue, 27 Sep 2016 11:09:37 GMT
http://git-wip-us.apache.org/repos/asf/ignite/blob/8032fc2c/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopClassLoader.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopClassLoader.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopClassLoader.java
deleted file mode 100644
index 2e0e271..0000000
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopClassLoader.java
+++ /dev/null
@@ -1,964 +0,0 @@
-/*
- * 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;
-
-import java.io.IOException;
-import java.io.InputStream;
-import java.net.URL;
-import java.net.URLClassLoader;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashSet;
-import java.util.Map;
-import java.util.Set;
-import java.util.UUID;
-import java.util.Vector;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-
-import org.apache.hadoop.util.NativeCodeLoader;
-import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.internal.processors.hadoop.v2.HadoopDaemon;
-import org.apache.ignite.internal.processors.hadoop.v2.HadoopShutdownHookManager;
-import org.apache.ignite.internal.util.ClassCache;
-import org.apache.ignite.internal.util.typedef.F;
-import org.apache.ignite.internal.util.typedef.internal.S;
-import org.apache.ignite.internal.util.typedef.internal.U;
-import org.jetbrains.annotations.Nullable;
-import org.jsr166.ConcurrentHashMap8;
-import org.objectweb.asm.AnnotationVisitor;
-import org.objectweb.asm.Attribute;
-import org.objectweb.asm.ClassReader;
-import org.objectweb.asm.ClassVisitor;
-import org.objectweb.asm.ClassWriter;
-import org.objectweb.asm.FieldVisitor;
-import org.objectweb.asm.Handle;
-import org.objectweb.asm.Label;
-import org.objectweb.asm.MethodVisitor;
-import org.objectweb.asm.Opcodes;
-import org.objectweb.asm.Type;
-import org.objectweb.asm.commons.Remapper;
-import org.objectweb.asm.commons.RemappingClassAdapter;
-
-/**
- * Class loader allowing explicitly load classes without delegation to parent class loader.
- * Also supports class parsing for finding dependencies which contain transitive dependencies
- * unavailable for parent.
- */
-public class HadoopClassLoader extends URLClassLoader implements ClassCache {
-    static {
-        // We are very parallel capable.
-        registerAsParallelCapable();
-    }
-
-    /** Name of the Hadoop daemon class. */
-    public static final String HADOOP_DAEMON_CLASS_NAME = "org.apache.hadoop.util.Daemon";
-
-    /** Name of libhadoop library. */
-    private static final String LIBHADOOP = "hadoop.";
-
-    /** */
-    private static final URLClassLoader APP_CLS_LDR = (URLClassLoader)HadoopClassLoader.class.getClassLoader();
-
-    /** */
-    private static final Collection<URL> appJars = F.asList(APP_CLS_LDR.getURLs());
-
-    /** */
-    private static volatile Collection<URL> hadoopJars;
-
-    /** */
-    private static final Map<String, Boolean> cache = new ConcurrentHashMap8<>();
-
-    /** */
-    private static final Map<String, byte[]> bytesCache = new ConcurrentHashMap8<>();
-
-    /** Class cache. */
-    private final ConcurrentMap<String, Class> cacheMap = new ConcurrentHashMap<>();
-
-    /** Diagnostic name of this class loader. */
-    @SuppressWarnings({"FieldCanBeLocal", "UnusedDeclaration"})
-    private final String name;
-
-    /** Native library names. */
-    private final String[] libNames;
-
-    /**
-     * Gets name for Job class loader. The name is specific for local node id.
-     * @param locNodeId The local node id.
-     * @return The class loader name.
-     */
-    public static String nameForJob(UUID locNodeId) {
-        return "hadoop-job-node-" + locNodeId.toString();
-    }
-
-    /**
-     * Gets name for the task class loader. Task class loader
-     * @param info The task info.
-     * @param prefix Get only prefix (without task type and number)
-     * @return The class loader name.
-     */
-    public static String nameForTask(HadoopTaskInfo info, boolean prefix) {
-        if (prefix)
-            return "hadoop-task-" + info.jobId() + "-";
-        else
-            return "hadoop-task-" + info.jobId() + "-" + info.type() + "-" + info.taskNumber();
-    }
-
-    /**
-     * Constructor.
-     *
-     * @param urls Urls.
-     * @param name Classloader name.
-     * @param libNames Optional additional native library names to be linked from parent classloader.
-     */
-    public HadoopClassLoader(URL[] urls, String name, @Nullable String[] libNames) {
-        super(addHadoopUrls(urls), APP_CLS_LDR);
-
-        assert !(getParent() instanceof HadoopClassLoader);
-
-        this.name = name;
-        this.libNames = libNames;
-
-        initializeNativeLibraries();
-    }
-
-    /**
-     * Workaround to load native Hadoop libraries. Java doesn't allow native libraries to be loaded from different
-     * classloaders. But we load Hadoop classes many times and one of these classes - {@code NativeCodeLoader} - tries
-     * to load the same native library over and over again.
-     * <p>
-     * To fix the problem, we force native library load in parent class loader and then "link" handle to this native
-     * library to our class loader. As a result, our class loader will think that the library is already loaded and will
-     * be able to link native methods.
-     *
-     * @see <a href="http://docs.oracle.com/javase/1.5.0/docs/guide/jni/spec/invocation.html#library_version">
-     *     JNI specification</a>
-     */
-    private void initializeNativeLibraries() {
-        try {
-            // This must trigger native library load.
-            Class.forName(NativeCodeLoader.class.getName(), true, APP_CLS_LDR);
-
-            final Vector<Object> curVector = U.field(this, "nativeLibraries");
-
-            ClassLoader ldr = APP_CLS_LDR;
-
-            while (ldr != null) {
-                Vector vector = U.field(ldr, "nativeLibraries");
-
-                for (Object lib : vector) {
-                    String name = U.field(lib, "name");
-
-                    boolean add = name.contains(LIBHADOOP);
-
-                    if (!add && libNames != null) {
-                        for (String libName : libNames) {
-                            if (libName != null && name.contains(libName)) {
-                                add = true;
-
-                                break;
-                            }
-                        }
-                    }
-
-                    if (add) {
-                        curVector.add(lib);
-
-                        return;
-                    }
-                }
-
-                ldr = ldr.getParent();
-            }
-        }
-        catch (Exception e) {
-            U.quietAndWarn(null, "Failed to initialize Hadoop native library " +
-                "(native Hadoop methods might not work properly): " + e);
-        }
-    }
-
-    /**
-     * Need to parse only Ignite Hadoop and IGFS classes.
-     *
-     * @param cls Class name.
-     * @return {@code true} if we need to check this class.
-     */
-    private static boolean isHadoopIgfs(String cls) {
-        String ignitePkgPrefix = "org.apache.ignite";
-
-        int len = ignitePkgPrefix.length();
-
-        return cls.startsWith(ignitePkgPrefix) && (
-            cls.indexOf("igfs.", len) != -1 ||
-            cls.indexOf(".fs.", len) != -1 ||
-            cls.indexOf("hadoop.", len) != -1);
-    }
-
-    /**
-     * @param cls Class name.
-     * @return {@code true} If this is Hadoop class.
-     */
-    private static boolean isHadoop(String cls) {
-        return cls.startsWith("org.apache.hadoop.");
-    }
-
-    /** {@inheritDoc} */
-    @Override protected Class<?> loadClass(String name, boolean resolve) throws ClassNotFoundException {
-        try {
-            if (isHadoop(name)) { // Always load Hadoop classes explicitly, since Hadoop can be available in App classpath.
-                if (name.endsWith(".util.ShutdownHookManager"))  // Dirty hack to get rid of Hadoop shutdown hooks.
-                    return loadFromBytes(name, HadoopShutdownHookManager.class.getName());
-                else if (name.equals(HADOOP_DAEMON_CLASS_NAME))
-                    // We replace this in order to be able to forcibly stop some daemon threads
-                    // that otherwise never stop (e.g. PeerCache runnables):
-                    return loadFromBytes(name, HadoopDaemon.class.getName());
-
-                return loadClassExplicitly(name, resolve);
-            }
-
-            if (isHadoopIgfs(name)) { // For Ignite Hadoop and IGFS classes we have to check if they depend on Hadoop.
-                Boolean hasDeps = cache.get(name);
-
-                if (hasDeps == null) {
-                    hasDeps = hasExternalDependencies(name);
-
-                    cache.put(name, hasDeps);
-                }
-
-                if (hasDeps)
-                    return loadClassExplicitly(name, resolve);
-            }
-
-            return super.loadClass(name, resolve);
-        }
-        catch (NoClassDefFoundError | ClassNotFoundException e) {
-            throw new ClassNotFoundException("Failed to load class: " + name, e);
-        }
-    }
-
-    /**
-     * @param name Name.
-     * @param replace Replacement.
-     * @return Class.
-     */
-    private Class<?> loadFromBytes(final String name, final String replace) {
-        synchronized (getClassLoadingLock(name)) {
-            // First, check if the class has already been loaded
-            Class c = findLoadedClass(name);
-
-            if (c != null)
-                return c;
-
-            byte[] bytes = bytesCache.get(name);
-
-            if (bytes == null) {
-                InputStream in = loadClassBytes(getParent(), replace);
-
-                ClassReader rdr;
-
-                try {
-                    rdr = new ClassReader(in);
-                }
-                catch (IOException e) {
-                    throw new RuntimeException(e);
-                }
-
-                ClassWriter w = new ClassWriter(Opcodes.ASM4);
-
-                rdr.accept(new RemappingClassAdapter(w, new Remapper() {
-                    /** */
-                    String replaceType = replace.replace('.', '/');
-
-                    /** */
-                    String nameType = name.replace('.', '/');
-
-                    @Override public String map(String type) {
-                        if (type.equals(replaceType))
-                            return nameType;
-
-                        return type;
-                    }
-                }), ClassReader.EXPAND_FRAMES);
-
-                bytes = w.toByteArray();
-
-                bytesCache.put(name, bytes);
-            }
-
-            return defineClass(name, bytes, 0, bytes.length);
-        }
-    }
-
-    /** {@inheritDoc} */
-    @Override public Class<?> getFromCache(String clsName) throws ClassNotFoundException {
-        Class<?> cls = cacheMap.get(clsName);
-
-        if (cls == null) {
-            Class old = cacheMap.putIfAbsent(clsName, cls = Class.forName(clsName, true, this));
-
-            if (old != null)
-                cls = old;
-        }
-
-        return cls;
-    }
-
-    /**
-     * @param name Class name.
-     * @param resolve Resolve class.
-     * @return Class.
-     * @throws ClassNotFoundException If failed.
-     */
-    private Class<?> loadClassExplicitly(String name, boolean resolve) throws ClassNotFoundException {
-        synchronized (getClassLoadingLock(name)) {
-            // First, check if the class has already been loaded
-            Class c = findLoadedClass(name);
-
-            if (c == null) {
-                long t1 = System.nanoTime();
-
-                c = findClass(name);
-
-                // this is the defining class loader; record the stats
-                sun.misc.PerfCounter.getFindClassTime().addElapsedTimeFrom(t1);
-                sun.misc.PerfCounter.getFindClasses().increment();
-            }
-
-            if (resolve)
-                resolveClass(c);
-
-            return c;
-        }
-    }
-
-    /**
-     * @param ldr Loader.
-     * @param clsName Class.
-     * @return Input stream.
-     */
-    @Nullable private InputStream loadClassBytes(ClassLoader ldr, String clsName) {
-        return ldr.getResourceAsStream(clsName.replace('.', '/') + ".class");
-    }
-
-    /**
-     * Check whether class has external dependencies on Hadoop.
-     *
-     * @param clsName Class name.
-     * @return {@code True} if class has external dependencies.
-     */
-    boolean hasExternalDependencies(String clsName) {
-        CollectingContext ctx = new CollectingContext();
-
-        ctx.annVisitor = new CollectingAnnotationVisitor(ctx);
-        ctx.mthdVisitor = new CollectingMethodVisitor(ctx, ctx.annVisitor);
-        ctx.fldVisitor = new CollectingFieldVisitor(ctx, ctx.annVisitor);
-        ctx.clsVisitor = new CollectingClassVisitor(ctx, ctx.annVisitor, ctx.mthdVisitor, ctx.fldVisitor);
-
-        return hasExternalDependencies(clsName, ctx);
-    }
-
-    /**
-     * Check whether class has external dependencies on Hadoop.
-     *
-     * @param clsName Class name.
-     * @param ctx Context.
-     * @return {@code true} If the class has external dependencies.
-     */
-    boolean hasExternalDependencies(String clsName, CollectingContext ctx) {
-        if (isHadoop(clsName)) // Hadoop must not be in classpath but Idea sucks, so filtering explicitly as external.
-            return true;
-
-        // Try to get from parent to check if the type accessible.
-        InputStream in = loadClassBytes(getParent(), clsName);
-
-        if (in == null) // The class is external itself, it must be loaded from this class loader.
-            return true;
-
-        if (!isHadoopIgfs(clsName)) // Other classes should not have external dependencies.
-            return false;
-
-        final ClassReader rdr;
-
-        try {
-            rdr = new ClassReader(in);
-        }
-        catch (IOException e) {
-            throw new RuntimeException("Failed to read class: " + clsName, e);
-        }
-
-        ctx.visited.add(clsName);
-
-        rdr.accept(ctx.clsVisitor, 0);
-
-        if (ctx.found) // We already know that we have dependencies, no need to check parent.
-            return true;
-
-        // Here we are known to not have any dependencies but possibly we have a parent which has them.
-        int idx = clsName.lastIndexOf('$');
-
-        if (idx == -1) // No parent class.
-            return false;
-
-        String parentCls = clsName.substring(0, idx);
-
-        if (ctx.visited.contains(parentCls))
-            return false;
-
-        Boolean res = cache.get(parentCls);
-
-        if (res == null)
-            res = hasExternalDependencies(parentCls, ctx);
-
-        return res;
-    }
-
-    /**
-     * @param name Class name.
-     * @return {@code true} If this is a valid class name.
-     */
-    private static boolean validateClassName(String name) {
-        int len = name.length();
-
-        if (len <= 1)
-            return false;
-
-        if (!Character.isJavaIdentifierStart(name.charAt(0)))
-            return false;
-
-        boolean hasDot = false;
-
-        for (int i = 1; i < len; i++) {
-            char c = name.charAt(i);
-
-            if (c == '.')
-                hasDot = true;
-            else if (!Character.isJavaIdentifierPart(c))
-                return false;
-        }
-
-        return hasDot;
-    }
-
-    /**
-     * @param urls URLs.
-     * @return URLs.
-     */
-    private static URL[] addHadoopUrls(URL[] urls) {
-        Collection<URL> hadoopJars;
-
-        try {
-            hadoopJars = hadoopUrls();
-        }
-        catch (IgniteCheckedException e) {
-            throw new RuntimeException(e);
-        }
-
-        ArrayList<URL> list = new ArrayList<>(hadoopJars.size() + appJars.size() + (urls == null ? 0 : urls.length));
-
-        list.addAll(appJars);
-        list.addAll(hadoopJars);
-
-        if (!F.isEmpty(urls))
-            list.addAll(F.asList(urls));
-
-        return list.toArray(new URL[list.size()]);
-    }
-
-    /**
-     * @return Collection of jar URLs.
-     * @throws IgniteCheckedException If failed.
-     */
-    public static Collection<URL> hadoopUrls() throws IgniteCheckedException {
-        Collection<URL> hadoopUrls = hadoopJars;
-
-        if (hadoopUrls != null)
-            return hadoopUrls;
-
-        synchronized (HadoopClassLoader.class) {
-            hadoopUrls = hadoopJars;
-
-            if (hadoopUrls != null)
-                return hadoopUrls;
-
-            try {
-                hadoopUrls = HadoopClasspathUtils.classpathForClassLoader();
-            }
-            catch (IOException e) {
-                throw new IgniteCheckedException("Failed to resolve Hadoop JAR locations: " + e.getMessage(), e);
-            }
-
-            hadoopJars = hadoopUrls;
-
-            return hadoopUrls;
-        }
-    }
-
-    /** {@inheritDoc} */
-    @Override public String toString() {
-        return S.toString(HadoopClassLoader.class, this);
-    }
-
-    /**
-     * Getter for name field.
-     */
-    public String name() {
-        return name;
-    }
-
-    /**
-     * Context for dependencies collection.
-     */
-    private class CollectingContext {
-        /** Visited classes. */
-        private final Set<String> visited = new HashSet<>();
-
-        /** Whether dependency found. */
-        private boolean found;
-
-        /** Annotation visitor. */
-        private AnnotationVisitor annVisitor;
-
-        /** Method visitor. */
-        private MethodVisitor mthdVisitor;
-
-        /** Field visitor. */
-        private FieldVisitor fldVisitor;
-
-        /** Class visitor. */
-        private ClassVisitor clsVisitor;
-
-        /**
-         * Processes a method descriptor
-         * @param methDesc The method desc String.
-         */
-        void onMethodsDesc(final String methDesc) {
-            // Process method return type:
-            onType(Type.getReturnType(methDesc));
-
-            if (found)
-                return;
-
-            // Process method argument types:
-            for (Type t: Type.getArgumentTypes(methDesc)) {
-                onType(t);
-
-                if (found)
-                    return;
-            }
-        }
-
-        /**
-         * Processes dependencies of a class.
-         *
-         * @param depCls The class name as dot-notated FQN.
-         */
-        void onClass(final String depCls) {
-            assert depCls.indexOf('/') == -1 : depCls; // class name should be fully converted to dot notation.
-            assert depCls.charAt(0) != 'L' : depCls;
-            assert validateClassName(depCls) : depCls;
-
-            if (depCls.startsWith("java.") || depCls.startsWith("javax.")) // Filter out platform classes.
-                return;
-
-            if (visited.contains(depCls))
-                return;
-
-            Boolean res = cache.get(depCls);
-
-            if (res == Boolean.TRUE || (res == null && hasExternalDependencies(depCls, this)))
-                found = true;
-        }
-
-        /**
-         * Analyses dependencies of given type.
-         *
-         * @param t The type to process.
-         */
-        void onType(Type t) {
-            if (t == null)
-                return;
-
-            int sort = t.getSort();
-
-            switch (sort) {
-                case Type.ARRAY:
-                    onType(t.getElementType());
-
-                    break;
-
-                case Type.OBJECT:
-                    onClass(t.getClassName());
-
-                    break;
-            }
-        }
-
-        /**
-         * Analyses dependencies of given object type.
-         *
-         * @param objType The object type to process.
-         */
-        void onInternalTypeName(String objType) {
-            if (objType == null)
-                return;
-
-            assert objType.length() > 1 : objType;
-
-            if (objType.charAt(0) == '[')
-                // handle array. In this case this is a type descriptor notation, like "[Ljava/lang/Object;"
-                onType(objType);
-            else {
-                assert objType.indexOf('.') == -1 : objType; // Must be slash-separated FQN.
-
-                String clsName = objType.replace('/', '.'); // Convert it to dot notation.
-
-                onClass(clsName); // Process.
-            }
-        }
-
-        /**
-         * Type description analyser.
-         *
-         * @param desc The description.
-         */
-        void onType(String desc) {
-            if (!F.isEmpty(desc)) {
-                if (desc.length() <= 1)
-                    return; // Optimization: filter out primitive types in early stage.
-
-                Type t = Type.getType(desc);
-
-                onType(t);
-            }
-        }
-    }
-
-    /**
-     * Annotation visitor.
-     */
-    private static class CollectingAnnotationVisitor extends AnnotationVisitor {
-        /** */
-        final CollectingContext ctx;
-
-        /**
-         * Annotation visitor.
-         *
-         * @param ctx The collector.
-         */
-        CollectingAnnotationVisitor(CollectingContext ctx) {
-            super(Opcodes.ASM4);
-
-            this.ctx = ctx;
-        }
-
-        /** {@inheritDoc} */
-        @Override public AnnotationVisitor visitAnnotation(String name, String desc) {
-            if (ctx.found)
-                return null;
-
-            ctx.onType(desc);
-
-            return this;
-        }
-
-        /** {@inheritDoc} */
-        @Override public void visitEnum(String name, String desc, String val) {
-            if (ctx.found)
-                return;
-
-            ctx.onType(desc);
-        }
-
-        /** {@inheritDoc} */
-        @Override public AnnotationVisitor visitArray(String name) {
-            return ctx.found ? null : this;
-        }
-
-        /** {@inheritDoc} */
-        @Override public void visit(String name, Object val) {
-            if (ctx.found)
-                return;
-
-            if (val instanceof Type)
-                ctx.onType((Type)val);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void visitEnd() {
-            // No-op.
-        }
-    }
-
-    /**
-     * Field visitor.
-     */
-    private static class CollectingFieldVisitor extends FieldVisitor {
-        /** Collector. */
-        private final CollectingContext ctx;
-
-        /** Annotation visitor. */
-        private final AnnotationVisitor av;
-
-        /**
-         * Constructor.
-         */
-        CollectingFieldVisitor(CollectingContext ctx, AnnotationVisitor av) {
-            super(Opcodes.ASM4);
-
-            this.ctx = ctx;
-            this.av = av;
-        }
-
-        /** {@inheritDoc} */
-        @Override public AnnotationVisitor visitAnnotation(String desc, boolean visible) {
-            if (ctx.found)
-                return null;
-
-            ctx.onType(desc);
-
-            return ctx.found ? null : av;
-        }
-
-        /** {@inheritDoc} */
-        @Override public void visitAttribute(Attribute attr) {
-            // No-op.
-        }
-
-        /** {@inheritDoc} */
-        @Override public void visitEnd() {
-            // No-op.
-        }
-    }
-
-    /**
-     * Class visitor.
-     */
-    private static class CollectingClassVisitor extends ClassVisitor {
-        /** Collector. */
-        private final CollectingContext ctx;
-
-        /** Annotation visitor. */
-        private final AnnotationVisitor av;
-
-        /** Method visitor. */
-        private final MethodVisitor mv;
-
-        /** Field visitor. */
-        private final FieldVisitor fv;
-
-        /**
-         * Constructor.
-         *
-         * @param ctx Collector.
-         * @param av Annotation visitor.
-         * @param mv Method visitor.
-         * @param fv Field visitor.
-         */
-        CollectingClassVisitor(CollectingContext ctx, AnnotationVisitor av, MethodVisitor mv, FieldVisitor fv) {
-            super(Opcodes.ASM4);
-
-            this.ctx = ctx;
-            this.av = av;
-            this.mv = mv;
-            this.fv = fv;
-        }
-
-        /** {@inheritDoc} */
-        @Override public void visit(int i, int i2, String name, String signature, String superName, String[] ifaces) {
-            if (ctx.found)
-                return;
-
-            ctx.onInternalTypeName(superName);
-
-            if (ctx.found)
-                return;
-
-            if (ifaces != null) {
-                for (String iface : ifaces) {
-                    ctx.onInternalTypeName(iface);
-
-                    if (ctx.found)
-                        return;
-                }
-            }
-        }
-
-        /** {@inheritDoc} */
-        @Override public AnnotationVisitor visitAnnotation(String desc, boolean visible) {
-            if (ctx.found)
-                return null;
-
-            ctx.onType(desc);
-
-            return ctx.found ? null : av;
-        }
-
-        /** {@inheritDoc} */
-        @Override public void visitInnerClass(String name, String outerName, String innerName, int i) {
-            if (ctx.found)
-                return;
-
-            ctx.onInternalTypeName(name);
-        }
-
-        /** {@inheritDoc} */
-        @Override public FieldVisitor visitField(int i, String name, String desc, String signature, Object val) {
-            if (ctx.found)
-                return null;
-
-            ctx.onType(desc);
-
-            return ctx.found ? null : fv;
-        }
-
-        /** {@inheritDoc} */
-        @Override public MethodVisitor visitMethod(int i, String name, String desc, String signature,
-            String[] exceptions) {
-            if (ctx.found)
-                return null;
-
-            ctx.onMethodsDesc(desc);
-
-            // Process declared method exceptions:
-            if (exceptions != null) {
-                for (String e : exceptions)
-                    ctx.onInternalTypeName(e);
-            }
-
-            return ctx.found ? null : mv;
-        }
-    }
-
-    /**
-     * Method visitor.
-     */
-    private static class CollectingMethodVisitor extends MethodVisitor {
-        /** Collector. */
-        private final CollectingContext ctx;
-
-        /** Annotation visitor. */
-        private final AnnotationVisitor av;
-
-        /**
-         * Constructor.
-         *
-         * @param ctx Collector.
-         * @param av Annotation visitor.
-         */
-        private CollectingMethodVisitor(CollectingContext ctx, AnnotationVisitor av) {
-            super(Opcodes.ASM4);
-
-            this.ctx = ctx;
-            this.av = av;
-        }
-
-        /** {@inheritDoc} */
-        @Override public AnnotationVisitor visitAnnotation(String desc, boolean visible) {
-            if (ctx.found)
-                return null;
-
-            ctx.onType(desc);
-
-            return ctx.found ? null : av;
-        }
-
-        /** {@inheritDoc} */
-        @Override public AnnotationVisitor visitParameterAnnotation(int i, String desc, boolean b) {
-            if (ctx.found)
-                return null;
-
-            ctx.onType(desc);
-
-            return ctx.found ? null : av;
-        }
-
-        /** {@inheritDoc} */
-        @Override public AnnotationVisitor visitAnnotationDefault() {
-            return ctx.found ? null : av;
-        }
-
-        /** {@inheritDoc} */
-        @Override public void visitFieldInsn(int opcode, String owner, String name, String desc) {
-            if (ctx.found)
-                return;
-
-            ctx.onInternalTypeName(owner);
-
-            if (ctx.found)
-                return;
-
-            ctx.onType(desc);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void visitInvokeDynamicInsn(String name, String desc, Handle bsm, Object... bsmArgs) {
-            // No-op.
-        }
-
-        /** {@inheritDoc} */
-        @Override public void visitFrame(int type, int nLoc, Object[] locTypes, int nStack, Object[] stackTypes) {
-            // No-op.
-        }
-
-        /** {@inheritDoc} */
-        @Override public void visitLocalVariable(String name, String desc, String signature, Label lb,
-            Label lb2, int i) {
-            if (ctx.found)
-                return;
-
-            ctx.onType(desc);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void visitMethodInsn(int i, String owner, String name, String desc) {
-            if (ctx.found)
-                return;
-
-            ctx.onInternalTypeName(owner);
-
-            if (ctx.found)
-                return;
-
-            ctx.onMethodsDesc(desc);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void visitMultiANewArrayInsn(String desc, int dim) {
-            if (ctx.found)
-                return;
-
-            ctx.onType(desc);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void visitTryCatchBlock(Label start, Label end, Label hndl, String typeStr) {
-            if (ctx.found)
-                return;
-
-            ctx.onInternalTypeName(typeStr);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void visitTypeInsn(int opcode, String type) {
-            if (ctx.found)
-                return;
-
-            ctx.onInternalTypeName(type);
-        }
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ignite/blob/8032fc2c/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopClasspathMain.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopClasspathMain.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopClasspathMain.java
deleted file mode 100644
index 4069496..0000000
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopClasspathMain.java
+++ /dev/null
@@ -1,44 +0,0 @@
-/*
- * 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;
-
-/**
- * Main class to compose Hadoop classpath depending on the environment.
- * This class is designed to be independent on any Ignite classes as possible.
- * Please make sure to pass the path separator character as the 1st parameter to the main method.
- */
-public class HadoopClasspathMain {
-    /**
-     * Main method to be executed from scripts. It prints the classpath to the standard output.
-     *
-     * @param args The 1st argument should be the path separator character (":" on Linux, ";" on Windows).
-     */
-    public static void main(String[] args) throws Exception {
-        if (args.length < 1)
-            throw new IllegalArgumentException("Path separator must be passed as the first argument.");
-
-        String separator = args[0];
-
-        StringBuilder sb = new StringBuilder();
-
-        for (String path : HadoopClasspathUtils.classpathForProcess())
-            sb.append(path).append(separator);
-
-        System.out.println(sb);
-    }
-}

http://git-wip-us.apache.org/repos/asf/ignite/blob/8032fc2c/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopClasspathUtils.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopClasspathUtils.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopClasspathUtils.java
deleted file mode 100644
index f5c2814..0000000
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopClasspathUtils.java
+++ /dev/null
@@ -1,461 +0,0 @@
-/*
- * 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;
-
-import java.io.File;
-import java.io.FilenameFilter;
-import java.io.IOException;
-import java.net.MalformedURLException;
-import java.net.URL;
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.LinkedList;
-import java.util.List;
-
-/**
- * Hadoop classpath utilities.
- */
-public class HadoopClasspathUtils {
-    /** Prefix directory. */
-    public static final String PREFIX = "HADOOP_PREFIX";
-
-    /** Hadoop home directory. */
-    public static final String HOME = "HADOOP_HOME";
-
-    /** Hadoop common directory. */
-    public static final String COMMON_HOME = "HADOOP_COMMON_HOME";
-
-    /** Hadoop HDFS directory. */
-    public static final String HDFS_HOME = "HADOOP_HDFS_HOME";
-
-    /** Hadoop mapred directory. */
-    public static final String MAPRED_HOME = "HADOOP_MAPRED_HOME";
-
-    /** Arbitrary additional dependencies. Compliant with standard Java classpath resolution. */
-    public static final String HADOOP_USER_LIBS = "HADOOP_USER_LIBS";
-
-    /** Empty string. */
-    private static final String EMPTY_STR = "";
-
-    /**
-     * Gets Hadoop class path as list of classpath elements for process.
-     *
-     * @return List of the class path elements.
-     * @throws IOException If failed.
-     */
-    public static List<String> classpathForProcess() throws IOException {
-        List<String> res = new ArrayList<>();
-
-        for (final SearchDirectory dir : classpathDirectories()) {
-            File[] files = dir.files();
-
-            if (dir.useWildcard()) {
-                if (files.length > 0)
-                    res.add(dir.absolutePath() + File.separator + '*');
-            }
-            else {
-                for (File file : files)
-                    res.add(file.getAbsolutePath());
-            }
-        }
-
-        return res;
-    }
-
-    /**
-     * Gets Hadoop class path as a list of URLs (for in-process class loader usage).
-     *
-     * @return List of class path URLs.
-     * @throws IOException If failed.
-     */
-    public static List<URL> classpathForClassLoader() throws IOException {
-        List<URL> res = new ArrayList<>();
-
-        for (SearchDirectory dir : classpathDirectories()) {
-            for (File file : dir.files()) {
-                try {
-                    res.add(file.toURI().toURL());
-                }
-                catch (MalformedURLException e) {
-                    throw new IOException("Failed to convert file path to URL: " + file.getPath());
-                }
-            }
-        }
-
-        return res;
-    }
-
-    /**
-     * Gets Hadoop locations.
-     *
-     * @return The locations as determined from the environment.
-     */
-    public static HadoopLocations locations() throws IOException {
-        // Query environment.
-        String hadoopHome = systemOrEnv(PREFIX, systemOrEnv(HOME, EMPTY_STR));
-
-        String commonHome = systemOrEnv(COMMON_HOME, EMPTY_STR);
-        String hdfsHome = systemOrEnv(HDFS_HOME, EMPTY_STR);
-        String mapredHome = systemOrEnv(MAPRED_HOME, EMPTY_STR);
-
-        // If any composite location is defined, use only them.
-        if (!isEmpty(commonHome) || !isEmpty(hdfsHome) || !isEmpty(mapredHome)) {
-            HadoopLocations res = new HadoopLocations(hadoopHome, commonHome, hdfsHome, mapredHome);
-
-            if (res.valid())
-                return res;
-            else
-                throw new IOException("Failed to resolve Hadoop classpath because some environment variables are " +
-                    "either undefined or point to nonexistent directories [" +
-                    "[env=" + COMMON_HOME + ", value=" + commonHome + ", exists=" + res.commonExists() + "], " +
-                    "[env=" + HDFS_HOME + ", value=" + hdfsHome + ", exists=" + res.hdfsExists() + "], " +
-                    "[env=" + MAPRED_HOME + ", value=" + mapredHome + ", exists=" + res.mapredExists() + "]]");
-        }
-        else if (!isEmpty(hadoopHome)) {
-            // All further checks will be based on HADOOP_HOME, so check for it's existence.
-            if (!exists(hadoopHome))
-                throw new IOException("Failed to resolve Hadoop classpath because " + HOME + " environment " +
-                    "variable points to nonexistent directory: " + hadoopHome);
-
-            // Probe Apache Hadoop.
-            HadoopLocations res = new HadoopLocations(
-                hadoopHome,
-                hadoopHome + "/share/hadoop/common",
-                hadoopHome + "/share/hadoop/hdfs",
-                hadoopHome + "/share/hadoop/mapreduce"
-            );
-
-            if (res.valid())
-                return res;
-
-            // Probe CDH.
-            res = new HadoopLocations(
-                hadoopHome,
-                hadoopHome,
-                hadoopHome + "/../hadoop-hdfs",
-                hadoopHome + "/../hadoop-mapreduce"
-            );
-
-            if (res.valid())
-                return res;
-
-            // Probe HDP.
-            res = new HadoopLocations(
-                hadoopHome,
-                hadoopHome,
-                hadoopHome + "/../hadoop-hdfs-client",
-                hadoopHome + "/../hadoop-mapreduce-client"
-            );
-
-            if (res.valid())
-                return res;
-
-            // Failed.
-            throw new IOException("Failed to resolve Hadoop classpath because " + HOME + " environment variable " +
-                "is either invalid or points to non-standard Hadoop distribution: " + hadoopHome);
-        }
-        else {
-            // Advise to set HADOOP_HOME only as this is preferred way to configure classpath.
-            throw new IOException("Failed to resolve Hadoop classpath (please define " + HOME + " environment " +
-                "variable and point it to your Hadoop distribution).");
-        }
-    }
-
-    /**
-     * Gets base directories to discover classpath elements in.
-     *
-     * @return Collection of directory and mask pairs.
-     * @throws IOException if a mandatory classpath location is not found.
-     */
-    private static Collection<SearchDirectory> classpathDirectories() throws IOException {
-        HadoopLocations loc = locations();
-
-        Collection<SearchDirectory> res = new ArrayList<>();
-
-        res.add(new SearchDirectory(new File(loc.common(), "lib"), AcceptAllDirectoryFilter.INSTANCE));
-        res.add(new SearchDirectory(new File(loc.hdfs(), "lib"), AcceptAllDirectoryFilter.INSTANCE));
-        res.add(new SearchDirectory(new File(loc.mapred(), "lib"), AcceptAllDirectoryFilter.INSTANCE));
-
-        res.add(new SearchDirectory(new File(loc.common()), new PrefixDirectoryFilter("hadoop-common-")));
-        res.add(new SearchDirectory(new File(loc.common()), new PrefixDirectoryFilter("hadoop-auth-")));
-
-        res.add(new SearchDirectory(new File(loc.hdfs()), new PrefixDirectoryFilter("hadoop-hdfs-")));
-
-        res.add(new SearchDirectory(new File(loc.mapred()),
-            new PrefixDirectoryFilter("hadoop-mapreduce-client-common")));
-        res.add(new SearchDirectory(new File(loc.mapred()),
-            new PrefixDirectoryFilter("hadoop-mapreduce-client-core")));
-
-        res.addAll(parseUserLibs());
-
-        return res;
-    }
-
-    /**
-     * Parse user libs.
-     *
-     * @return Parsed libs search patterns.
-     * @throws IOException If failed.
-     */
-    static Collection<SearchDirectory> parseUserLibs() throws IOException {
-        return parseUserLibs(systemOrEnv(HADOOP_USER_LIBS, null));
-    }
-
-    /**
-     * Parse user libs.
-     *
-     * @param str String.
-     * @return Result.
-     * @throws IOException If failed.
-     */
-    static Collection<SearchDirectory> parseUserLibs(String str) throws IOException {
-        Collection<SearchDirectory> res = new LinkedList<>();
-
-        if (!isEmpty(str)) {
-            String[] tokens = normalize(str).split(File.pathSeparator);
-
-            for (String token : tokens) {
-                // Skip empty tokens.
-                if (isEmpty(token))
-                    continue;
-
-                File file = new File(token);
-                File dir = file.getParentFile();
-
-                if (token.endsWith("*")) {
-                    assert dir != null;
-
-                    res.add(new SearchDirectory(dir, AcceptAllDirectoryFilter.INSTANCE, false));
-                }
-                else {
-                    // Met "/" or "C:\" pattern, nothing to do with it.
-                    if (dir == null)
-                        continue;
-
-                    res.add(new SearchDirectory(dir, new ExactDirectoryFilter(file.getName()), false));
-                }
-            }
-        }
-
-        return res;
-    }
-
-    /**
-     * Get system property or environment variable with the given name.
-     *
-     * @param name Variable name.
-     * @param dflt Default value.
-     * @return Value.
-     */
-    private static String systemOrEnv(String name, String dflt) {
-        String res = System.getProperty(name);
-
-        if (res == null)
-            res = System.getenv(name);
-
-        return res != null ? res : dflt;
-    }
-
-    /**
-     * Answers if the given path denotes existing directory.
-     *
-     * @param path The directory path.
-     * @return {@code True} if the given path denotes an existing directory.
-     */
-    public static boolean exists(String path) {
-        if (path == null)
-            return false;
-
-        Path p = Paths.get(path);
-
-        return Files.exists(p) && Files.isDirectory(p) && Files.isReadable(p);
-    }
-
-    /**
-     * Check if string is empty.
-     *
-     * @param val Value.
-     * @return {@code True} if empty.
-     */
-    private static boolean isEmpty(String val) {
-        return val == null || val.isEmpty();
-    }
-
-    /**
-     * NOramlize the string.
-     *
-     * @param str String.
-     * @return Normalized string.
-     */
-    private static String normalize(String str) {
-        assert str != null;
-
-        return str.trim().toLowerCase();
-    }
-
-    /**
-     * Simple pair-like structure to hold directory name and a mask assigned to it.
-     */
-    static class SearchDirectory {
-        /** File. */
-        private final File dir;
-
-        /** Filter. */
-        private final DirectoryFilter filter;
-
-        /** Whether directory must exist. */
-        private final boolean strict;
-
-        /**
-         * Constructor for directory search with strict rule.
-         *
-         * @param dir Directory.
-         * @param filter Filter.
-         * @throws IOException If failed.
-         */
-        private SearchDirectory(File dir, DirectoryFilter filter) throws IOException {
-            this(dir, filter, true);
-        }
-
-        /**
-         * Constructor.
-         *
-         * @param dir Directory.
-         * @param filter Filter.
-         * @param strict Whether directory must exist.
-         * @throws IOException If failed.
-         */
-        private SearchDirectory(File dir, DirectoryFilter filter, boolean strict) throws IOException {
-            this.dir = dir;
-            this.filter = filter;
-            this.strict = strict;
-
-            if (strict && !exists(dir.getAbsolutePath()))
-                throw new IOException("Directory cannot be read: " + dir.getAbsolutePath());
-        }
-
-        /**
-         * @return Absolute path.
-         */
-        String absolutePath() {
-            return dir.getAbsolutePath();
-        }
-
-        /**
-         * @return Child files.
-         */
-        File[] files() throws IOException {
-            File[] files = dir.listFiles(new FilenameFilter() {
-                @Override public boolean accept(File dir, String name) {
-                    return filter.test(name);
-                }
-            });
-
-            if (files == null) {
-                if (strict)
-                    throw new IOException("Failed to get directory files [dir=" + dir + ']');
-                else
-                    return new File[0];
-            }
-            else
-                return files;
-        }
-
-        /**
-         * @return {@code True} if wildcard can be used.
-         */
-        boolean useWildcard() {
-            return filter instanceof AcceptAllDirectoryFilter;
-        }
-    }
-
-    /**
-     * Directory filter interface.
-     */
-    static interface DirectoryFilter {
-        /**
-         * Test if file with this name should be included.
-         *
-         * @param name File name.
-         * @return {@code True} if passed.
-         */
-        public boolean test(String name);
-    }
-
-    /**
-     * Filter to accept all files.
-     */
-    static class AcceptAllDirectoryFilter implements DirectoryFilter {
-        /** Singleton instance. */
-        public static final AcceptAllDirectoryFilter INSTANCE = new AcceptAllDirectoryFilter();
-
-        /** {@inheritDoc} */
-        @Override public boolean test(String name) {
-            return true;
-        }
-    }
-
-    /**
-     * Filter which uses prefix to filter files.
-     */
-    static class PrefixDirectoryFilter implements DirectoryFilter {
-        /** Prefix. */
-        private final String prefix;
-
-        /**
-         * Constructor.
-         *
-         * @param prefix Prefix.
-         */
-        public PrefixDirectoryFilter(String prefix) {
-            assert prefix != null;
-
-            this.prefix = normalize(prefix);
-        }
-
-        /** {@inheritDoc} */
-        @Override public boolean test(String name) {
-            return normalize(name).startsWith(prefix);
-        }
-    }
-
-    /**
-     * Filter which uses exact comparison.
-     */
-    static class ExactDirectoryFilter implements DirectoryFilter {
-        /** Name. */
-        private final String name;
-
-        /**
-         * Constructor.
-         *
-         * @param name Name.
-         */
-        public ExactDirectoryFilter(String name) {
-            this.name = normalize(name);
-        }
-
-        /** {@inheritDoc} */
-        @Override public boolean test(String name) {
-            return normalize(name).equals(this.name);
-        }
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ignite/blob/8032fc2c/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopCommonUtils.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopCommonUtils.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopCommonUtils.java
new file mode 100644
index 0000000..37af147
--- /dev/null
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopCommonUtils.java
@@ -0,0 +1,154 @@
+/*
+ * 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;
+
+import org.jetbrains.annotations.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.TreeSet;
+
+/**
+ * Common Hadoop utility methods which do not depend on Hadoop API.
+ */
+public class HadoopCommonUtils {
+    /** Job class name. */
+    public static final String JOB_CLS_NAME = "org.apache.ignite.internal.processors.hadoop.impl.v2.HadoopV2Job";
+
+    /** Property to store timestamp of new job id request. */
+    public static final String REQ_NEW_JOBID_TS_PROPERTY = "ignite.job.requestNewIdTs";
+
+    /** Property to store timestamp of response of new job id request. */
+    public static final String RESPONSE_NEW_JOBID_TS_PROPERTY = "ignite.job.responseNewIdTs";
+
+    /** Property to store timestamp of job submission. */
+    public static final String JOB_SUBMISSION_START_TS_PROPERTY = "ignite.job.submissionStartTs";
+
+    /** Property to set custom writer of job statistics. */
+    public static final String JOB_COUNTER_WRITER_PROPERTY = "ignite.counters.writer";
+
+    /**
+     * Sort input splits by length.
+     *
+     * @param splits Splits.
+     * @return Sorted splits.
+     */
+    public static List<HadoopInputSplit> sortInputSplits(Collection<HadoopInputSplit> splits) {
+        int id = 0;
+
+        TreeSet<SplitSortWrapper> sortedSplits = new TreeSet<>();
+
+        for (HadoopInputSplit split : splits) {
+            long len = split instanceof HadoopFileBlock ? ((HadoopFileBlock)split).length() : 0;
+
+            sortedSplits.add(new SplitSortWrapper(id++, split, len));
+        }
+
+        ArrayList<HadoopInputSplit> res = new ArrayList<>(sortedSplits.size());
+
+        for (SplitSortWrapper sortedSplit : sortedSplits)
+            res.add(sortedSplit.split);
+
+        return res;
+    }
+
+    /**
+     * Set context class loader.
+     *
+     * @param newLdr New class loader.
+     * @return Old class loader.
+     */
+    @Nullable public static ClassLoader setContextClassLoader(@Nullable ClassLoader newLdr) {
+        ClassLoader oldLdr = Thread.currentThread().getContextClassLoader();
+
+        if (newLdr != oldLdr)
+            Thread.currentThread().setContextClassLoader(newLdr);
+
+        return oldLdr;
+    }
+
+    /**
+     * Restore context class loader.
+     *
+     * @param oldLdr Original class loader.
+     */
+    public static void restoreContextClassLoader(@Nullable ClassLoader oldLdr) {
+        ClassLoader newLdr = Thread.currentThread().getContextClassLoader();
+
+        if (newLdr != oldLdr)
+            Thread.currentThread().setContextClassLoader(oldLdr);
+    }
+
+    /**
+     * Split wrapper for sorting.
+     */
+    private static class SplitSortWrapper implements Comparable<SplitSortWrapper> {
+        /** Unique ID. */
+        private final int id;
+
+        /** Split. */
+        private final HadoopInputSplit split;
+
+        /** Split length. */
+        private final long len;
+
+        /**
+         * Constructor.
+         *
+         * @param id Unique ID.
+         * @param split Split.
+         * @param len Split length.
+         */
+        public SplitSortWrapper(int id, HadoopInputSplit split, long len) {
+            this.id = id;
+            this.split = split;
+            this.len = len;
+        }
+
+        /** {@inheritDoc} */
+        @SuppressWarnings("NullableProblems")
+        @Override public int compareTo(SplitSortWrapper other) {
+            long res = len - other.len;
+
+            if (res > 0)
+                return -1;
+            else if (res < 0)
+                return 1;
+            else
+                return id - other.id;
+        }
+
+        /** {@inheritDoc} */
+        @Override public int hashCode() {
+            return id;
+        }
+
+        /** {@inheritDoc} */
+        @Override public boolean equals(Object obj) {
+            return obj instanceof SplitSortWrapper && id == ((SplitSortWrapper)obj).id;
+        }
+    }
+
+    /**
+     * Private constructor.
+     */
+    private HadoopCommonUtils() {
+        // No-op.
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/8032fc2c/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopContext.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopContext.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopContext.java
index 42a3d72..4326ad2 100644
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopContext.java
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopContext.java
@@ -24,7 +24,6 @@ import java.util.UUID;
 import org.apache.ignite.cluster.ClusterNode;
 import org.apache.ignite.configuration.HadoopConfiguration;
 import org.apache.ignite.internal.GridKernalContext;
-import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
 import org.apache.ignite.internal.processors.hadoop.jobtracker.HadoopJobMetadata;
 import org.apache.ignite.internal.processors.hadoop.jobtracker.HadoopJobTracker;
 import org.apache.ignite.internal.processors.hadoop.shuffle.HadoopShuffle;

http://git-wip-us.apache.org/repos/asf/ignite/blob/8032fc2c/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopDefaultJobInfo.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopDefaultJobInfo.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopDefaultJobInfo.java
deleted file mode 100644
index 1382c1f..0000000
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopDefaultJobInfo.java
+++ /dev/null
@@ -1,156 +0,0 @@
-/*
- * 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;
-
-import java.io.Externalizable;
-import java.io.IOException;
-import java.io.ObjectInput;
-import java.io.ObjectOutput;
-import java.lang.reflect.Constructor;
-import java.util.HashMap;
-import java.util.Map;
-import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.IgniteLogger;
-import org.apache.ignite.internal.util.typedef.internal.U;
-import org.jetbrains.annotations.Nullable;
-
-/**
- * Hadoop job info based on default Hadoop configuration.
- */
-public class HadoopDefaultJobInfo implements HadoopJobInfo, Externalizable {
-    /** */
-    private static final long serialVersionUID = 5489900236464999951L;
-
-    /** {@code true} If job has combiner. */
-    private boolean hasCombiner;
-
-    /** Number of reducers configured for job. */
-    private int numReduces;
-
-    /** Configuration. */
-    private Map<String,String> props = new HashMap<>();
-
-    /** Job name. */
-    private String jobName;
-
-    /** User name. */
-    private String user;
-
-    /**
-     * Default constructor required by {@link Externalizable}.
-     */
-    public HadoopDefaultJobInfo() {
-        // No-op.
-    }
-
-    /**
-     * Constructor.
-     *
-     * @param jobName Job name.
-     * @param user User name.
-     * @param hasCombiner {@code true} If job has combiner.
-     * @param numReduces Number of reducers configured for job.
-     * @param props All other properties of the job.
-     */
-    public HadoopDefaultJobInfo(String jobName, String user, boolean hasCombiner, int numReduces,
-        Map<String, String> props) {
-        this.jobName = jobName;
-        this.user = user;
-        this.hasCombiner = hasCombiner;
-        this.numReduces = numReduces;
-        this.props = props;
-    }
-
-    /** {@inheritDoc} */
-    @Nullable @Override public String property(String name) {
-        return props.get(name);
-    }
-
-    /** {@inheritDoc} */
-    @Override public HadoopJob createJob(Class<? extends HadoopJob> jobCls, HadoopJobId jobId, IgniteLogger log,
-        @Nullable String[] libNames) throws IgniteCheckedException {
-        assert jobCls != null;
-
-        try {
-            Constructor<? extends HadoopJob> constructor = jobCls.getConstructor(HadoopJobId.class,
-                HadoopDefaultJobInfo.class, IgniteLogger.class, String[].class);
-
-            return constructor.newInstance(jobId, this, log, libNames);
-        }
-        catch (Throwable t) {
-            if (t instanceof Error)
-                throw (Error)t;
-            
-            throw new IgniteCheckedException(t);
-        }
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean hasCombiner() {
-        return hasCombiner;
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean hasReducer() {
-        return reducers() > 0;
-    }
-
-    /** {@inheritDoc} */
-    @Override public int reducers() {
-        return numReduces;
-    }
-
-    /** {@inheritDoc} */
-    @Override public String jobName() {
-        return jobName;
-    }
-
-    /** {@inheritDoc} */
-    @Override public String user() {
-        return user;
-    }
-
-    /** {@inheritDoc} */
-    @Override public void writeExternal(ObjectOutput out) throws IOException {
-        U.writeString(out, jobName);
-        U.writeString(out, user);
-
-        out.writeBoolean(hasCombiner);
-        out.writeInt(numReduces);
-
-        U.writeStringMap(out, props);
-    }
-
-    /** {@inheritDoc} */
-    @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
-        jobName = U.readString(in);
-        user = U.readString(in);
-
-        hasCombiner = in.readBoolean();
-        numReduces = in.readInt();
-
-        props = U.readStringMap(in);
-    }
-
-    /**
-     * @return Properties of the job.
-     */
-    public Map<String, String> properties() {
-        return props;
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ignite/blob/8032fc2c/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopExternalSplit.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopExternalSplit.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopExternalSplit.java
new file mode 100644
index 0000000..bd767b3
--- /dev/null
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopExternalSplit.java
@@ -0,0 +1,88 @@
+/*
+ * 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;
+
+import java.io.Externalizable;
+import java.io.IOException;
+import java.io.ObjectInput;
+import java.io.ObjectOutput;
+
+/**
+ * Split serialized in external file.
+ */
+public class HadoopExternalSplit extends HadoopInputSplit {
+    /** */
+    private static final long serialVersionUID = 0L;
+
+    /** */
+    private long off;
+
+    /**
+     * For {@link Externalizable}.
+     */
+    public HadoopExternalSplit() {
+        // No-op.
+    }
+
+    /**
+     * @param hosts Hosts.
+     * @param off Offset of this split in external file.
+     */
+    public HadoopExternalSplit(String[] hosts, long off) {
+        assert off >= 0 : off;
+        assert hosts != null;
+
+        this.hosts = hosts;
+        this.off = off;
+    }
+
+    /**
+     * @return Offset of this input split in external file.
+     */
+    public long offset() {
+        return off;
+    }
+
+    /** {@inheritDoc} */
+    @Override public void writeExternal(ObjectOutput out) throws IOException {
+        out.writeLong(off);
+    }
+
+    /** {@inheritDoc} */
+    @Override public void readExternal(ObjectInput in) throws IOException, ClassNotFoundException {
+        off = in.readLong();
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean equals(Object o) {
+        if (this == o)
+            return true;
+
+        if (o == null || getClass() != o.getClass())
+            return false;
+
+        HadoopExternalSplit that = (HadoopExternalSplit) o;
+
+        return off == that.off;
+    }
+
+    /** {@inheritDoc} */
+    @Override public int hashCode() {
+        return (int)(off ^ (off >>> 32));
+    }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ignite/blob/8032fc2c/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopHelperImpl.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopHelperImpl.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopHelperImpl.java
new file mode 100644
index 0000000..71bb8a4
--- /dev/null
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopHelperImpl.java
@@ -0,0 +1,120 @@
+/*
+ * 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;
+
+import org.apache.ignite.internal.GridKernalContext;
+import org.jetbrains.annotations.Nullable;
+import org.objectweb.asm.ClassReader;
+import org.objectweb.asm.ClassWriter;
+import org.objectweb.asm.Opcodes;
+import org.objectweb.asm.commons.Remapper;
+import org.objectweb.asm.commons.RemappingClassAdapter;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+/**
+ * Utility methods for Hadoop classloader required to avoid direct 3rd-party dependencies in class loader.
+ */
+public class HadoopHelperImpl implements HadoopHelper {
+    /** Kernal context. */
+    private final GridKernalContext ctx;
+
+    /** Common class loader. */
+    private volatile HadoopClassLoader ldr;
+
+    /**
+     * Default constructor.
+     */
+    public HadoopHelperImpl() {
+        this(null);
+    }
+
+    /**
+     * Constructor.
+     *
+     * @param ctx Kernal context.
+     */
+    public HadoopHelperImpl(GridKernalContext ctx) {
+        this.ctx = ctx;
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean isNoOp() {
+        return false;
+    }
+
+    /** {@inheritDoc} */
+    @Override public HadoopClassLoader commonClassLoader() {
+        HadoopClassLoader res = ldr;
+
+        if (res == null) {
+            synchronized (this) {
+                res = ldr;
+
+                if (res == null) {
+                    String[] libNames = null;
+
+                    if (ctx != null && ctx.config().getHadoopConfiguration() != null)
+                        libNames = ctx.config().getHadoopConfiguration().getNativeLibraryNames();
+
+                    res = new HadoopClassLoader(null, "hadoop-common", libNames, this);
+
+                    ldr = res;
+                }
+            }
+        }
+
+        return res;
+    }
+
+    /** {@inheritDoc} */
+    @Override public byte[] loadReplace(InputStream in, final String originalName, final String replaceName) {
+        ClassReader rdr;
+
+        try {
+            rdr = new ClassReader(in);
+        }
+        catch (IOException e) {
+            throw new RuntimeException(e);
+        }
+
+        ClassWriter w = new ClassWriter(Opcodes.ASM4);
+
+        rdr.accept(new RemappingClassAdapter(w, new Remapper() {
+            /** */
+            String replaceType = replaceName.replace('.', '/');
+
+            /** */
+            String nameType = originalName.replace('.', '/');
+
+            @Override public String map(String type) {
+                if (type.equals(replaceType))
+                    return nameType;
+
+                return type;
+            }
+        }), ClassReader.EXPAND_FRAMES);
+
+        return w.toByteArray();
+    }
+
+    /** {@inheritDoc} */
+    @Override @Nullable public InputStream loadClassBytes(ClassLoader ldr, String clsName) {
+        return ldr.getResourceAsStream(clsName.replace('.', '/') + ".class");
+    }
+}

http://git-wip-us.apache.org/repos/asf/ignite/blob/8032fc2c/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopLocations.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopLocations.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopLocations.java
deleted file mode 100644
index a90007f..0000000
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopLocations.java
+++ /dev/null
@@ -1,123 +0,0 @@
-/*
- * 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;
-
-/**
- * Simple structure to hold Hadoop directory locations.
- */
-public class HadoopLocations {
-    /** Hadoop home. */
-    private final String home;
-
-    /** Common home. */
-    private final String common;
-
-    /** HDFS home. */
-    private final String hdfs;
-
-    /** Mapred home. */
-    private final String mapred;
-
-    /** Whether common home exists. */
-    private final boolean commonExists;
-
-    /** Whether HDFS home exists. */
-    private final boolean hdfsExists;
-
-    /** Whether mapred home exists. */
-    private final boolean mapredExists;
-
-    /**
-     * Constructor.
-     *
-     * @param home Hadoop home.
-     * @param common Common home.
-     * @param hdfs HDFS home.
-     * @param mapred Mapred home.
-     */
-    public HadoopLocations(String home, String common, String hdfs, String mapred) {
-        assert common != null && hdfs != null && mapred != null;
-
-        this.home = home;
-        this.common = common;
-        this.hdfs = hdfs;
-        this.mapred = mapred;
-
-        commonExists = HadoopClasspathUtils.exists(common);
-        hdfsExists = HadoopClasspathUtils.exists(hdfs);
-        mapredExists = HadoopClasspathUtils.exists(mapred);
-    }
-
-    /**
-     * @return Hadoop home.
-     */
-    public String home() {
-        return home;
-    }
-
-    /**
-     * @return Common home.
-     */
-    public String common() {
-        return common;
-    }
-
-    /**
-     * @return HDFS home.
-     */
-    public String hdfs() {
-        return hdfs;
-    }
-
-    /**
-     * @return Mapred home.
-     */
-    public String mapred() {
-        return mapred;
-    }
-
-    /**
-     * @return Whether common home exists.
-     */
-    public boolean commonExists() {
-        return commonExists;
-    }
-
-    /**
-     * @return Whether HDFS home exists.
-     */
-    public boolean hdfsExists() {
-        return hdfsExists;
-    }
-
-    /**
-     * @return Whether mapred home exists.
-     */
-    public boolean mapredExists() {
-        return mapredExists;
-    }
-
-    /**
-     * Whether all required directories exists.
-     *
-     * @return {@code True} if exists.
-     */
-    public boolean valid() {
-        return commonExists && hdfsExists && mapredExists;
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ignite/blob/8032fc2c/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceCounterGroup.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceCounterGroup.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceCounterGroup.java
deleted file mode 100644
index 4e03e17..0000000
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceCounterGroup.java
+++ /dev/null
@@ -1,123 +0,0 @@
-/*
- * 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;
-
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-import java.util.Iterator;
-import org.apache.hadoop.mapreduce.Counter;
-import org.apache.hadoop.mapreduce.CounterGroup;
-import org.apache.hadoop.mapreduce.counters.CounterGroupBase;
-
-/**
- * Hadoop +counter group adapter.
- */
-class HadoopMapReduceCounterGroup implements CounterGroup {
-    /** Counters. */
-    private final HadoopMapReduceCounters cntrs;
-
-    /** Group name. */
-    private final String name;
-
-    /**
-     * Creates new instance.
-     *
-     * @param cntrs Client counters instance.
-     * @param name Group name.
-     */
-    HadoopMapReduceCounterGroup(HadoopMapReduceCounters cntrs, String name) {
-        this.cntrs = cntrs;
-        this.name = name;
-    }
-
-    /** {@inheritDoc} */
-    @Override public String getName() {
-        return name;
-    }
-
-    /** {@inheritDoc} */
-    @Override public String getDisplayName() {
-        return name;
-    }
-
-    /** {@inheritDoc} */
-    @Override public void setDisplayName(String displayName) {
-        // No-op.
-    }
-
-    /** {@inheritDoc} */
-    @Override public void addCounter(Counter counter) {
-        addCounter(counter.getName(), counter.getDisplayName(), 0);
-    }
-
-    /** {@inheritDoc} */
-    @Override public Counter addCounter(String name, String displayName, long value) {
-        final Counter counter = cntrs.findCounter(this.name, name);
-
-        counter.setValue(value);
-
-        return counter;
-    }
-
-    /** {@inheritDoc} */
-    @Override public Counter findCounter(String counterName, String displayName) {
-        return cntrs.findCounter(name, counterName);
-    }
-
-    /** {@inheritDoc} */
-    @Override public Counter findCounter(String counterName, boolean create) {
-        return cntrs.findCounter(name, counterName, create);
-    }
-
-    /** {@inheritDoc} */
-    @Override public Counter findCounter(String counterName) {
-        return cntrs.findCounter(name, counterName);
-    }
-
-    /** {@inheritDoc} */
-    @Override public int size() {
-        return cntrs.groupSize(name);
-    }
-
-    /** {@inheritDoc} */
-    @Override public void incrAllCounters(CounterGroupBase<Counter> rightGroup) {
-        for (final Counter counter : rightGroup)
-            cntrs.findCounter(name, counter.getName()).increment(counter.getValue());
-    }
-
-    /** {@inheritDoc} */
-    @Override public CounterGroupBase<Counter> getUnderlyingGroup() {
-        return this;
-    }
-
-    /** {@inheritDoc} */
-    @Override public Iterator<Counter> iterator() {
-        return cntrs.iterateGroup(name);
-    }
-
-    /** {@inheritDoc} */
-    @Override public void write(DataOutput out) throws IOException {
-        throw new UnsupportedOperationException("not implemented");
-    }
-
-    /** {@inheritDoc} */
-    @Override public void readFields(DataInput in) throws IOException {
-        throw new UnsupportedOperationException("not implemented");
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ignite/blob/8032fc2c/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceCounters.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceCounters.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceCounters.java
deleted file mode 100644
index 57a853f..0000000
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceCounters.java
+++ /dev/null
@@ -1,228 +0,0 @@
-/*
- * 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;
-
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.NoSuchElementException;
-import org.apache.hadoop.mapreduce.Counter;
-import org.apache.hadoop.mapreduce.CounterGroup;
-import org.apache.hadoop.mapreduce.Counters;
-import org.apache.hadoop.mapreduce.FileSystemCounter;
-import org.apache.hadoop.mapreduce.counters.AbstractCounters;
-import org.apache.hadoop.mapreduce.counters.Limits;
-import org.apache.ignite.internal.processors.hadoop.counter.HadoopCounter;
-import org.apache.ignite.internal.processors.hadoop.counter.HadoopLongCounter;
-import org.apache.ignite.internal.processors.hadoop.v2.HadoopV2Counter;
-import org.apache.ignite.internal.util.typedef.T2;
-
-/**
- * Hadoop counters adapter.
- */
-public class HadoopMapReduceCounters extends Counters {
-    /** */
-    private final Map<T2<String,String>,HadoopLongCounter> cntrs = new HashMap<>();
-
-    /**
-     * Creates new instance based on given counters.
-     *
-     * @param cntrs Counters to adapt.
-     */
-    public HadoopMapReduceCounters(org.apache.ignite.internal.processors.hadoop.counter.HadoopCounters cntrs) {
-        for (HadoopCounter cntr : cntrs.all())
-            if (cntr instanceof HadoopLongCounter)
-                this.cntrs.put(new T2<>(cntr.group(), cntr.name()), (HadoopLongCounter) cntr);
-    }
-
-    /** {@inheritDoc} */
-    @Override public synchronized CounterGroup addGroup(CounterGroup grp) {
-        return addGroup(grp.getName(), grp.getDisplayName());
-    }
-
-    /** {@inheritDoc} */
-    @Override public CounterGroup addGroup(String name, String displayName) {
-        return new HadoopMapReduceCounterGroup(this, name);
-    }
-
-    /** {@inheritDoc} */
-    @Override public Counter findCounter(String grpName, String cntrName) {
-        return findCounter(grpName, cntrName, true);
-    }
-
-    /** {@inheritDoc} */
-    @Override public synchronized Counter findCounter(Enum<?> key) {
-        return findCounter(key.getDeclaringClass().getName(), key.name(), true);
-    }
-
-    /** {@inheritDoc} */
-    @Override public synchronized Counter findCounter(String scheme, FileSystemCounter key) {
-        return findCounter(String.format("FileSystem Counter (%s)", scheme), key.name());
-    }
-
-    /** {@inheritDoc} */
-    @Override public synchronized Iterable<String> getGroupNames() {
-        Collection<String> res = new HashSet<>();
-
-        for (HadoopCounter counter : cntrs.values())
-            res.add(counter.group());
-
-        return res;
-    }
-
-    /** {@inheritDoc} */
-    @Override public Iterator<CounterGroup> iterator() {
-        final Iterator<String> iter = getGroupNames().iterator();
-
-        return new Iterator<CounterGroup>() {
-            @Override public boolean hasNext() {
-                return iter.hasNext();
-            }
-
-            @Override public CounterGroup next() {
-                if (!hasNext())
-                    throw new NoSuchElementException();
-
-                return new HadoopMapReduceCounterGroup(HadoopMapReduceCounters.this, iter.next());
-            }
-
-            @Override public void remove() {
-                throw new UnsupportedOperationException("not implemented");
-            }
-        };
-    }
-
-    /** {@inheritDoc} */
-    @Override public synchronized CounterGroup getGroup(String grpName) {
-        return new HadoopMapReduceCounterGroup(this, grpName);
-    }
-
-    /** {@inheritDoc} */
-    @Override public synchronized int countCounters() {
-        return cntrs.size();
-    }
-
-    /** {@inheritDoc} */
-    @Override public synchronized void write(DataOutput out) throws IOException {
-        throw new UnsupportedOperationException("not implemented");
-    }
-
-    /** {@inheritDoc} */
-    @Override public synchronized void readFields(DataInput in) throws IOException {
-        throw new UnsupportedOperationException("not implemented");
-    }
-
-    /** {@inheritDoc} */
-    @Override public synchronized void incrAllCounters(AbstractCounters<Counter, CounterGroup> other) {
-        for (CounterGroup group : other) {
-            for (Counter counter : group) {
-                findCounter(group.getName(), counter.getName()).increment(counter.getValue());
-            }
-        }
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean equals(Object genericRight) {
-        if (!(genericRight instanceof HadoopMapReduceCounters))
-            return false;
-
-        return cntrs.equals(((HadoopMapReduceCounters) genericRight).cntrs);
-    }
-
-    /** {@inheritDoc} */
-    @Override public int hashCode() {
-        return cntrs.hashCode();
-    }
-
-    /** {@inheritDoc} */
-    @Override public void setWriteAllCounters(boolean snd) {
-        // No-op.
-    }
-
-    /** {@inheritDoc} */
-    @Override public boolean getWriteAllCounters() {
-        return true;
-    }
-
-    /** {@inheritDoc} */
-    @Override public Limits limits() {
-        return null;
-    }
-
-    /**
-     * Returns size of a group.
-     *
-     * @param grpName Name of the group.
-     * @return amount of counters in the given group.
-     */
-    public int groupSize(String grpName) {
-        int res = 0;
-
-        for (HadoopCounter counter : cntrs.values()) {
-            if (grpName.equals(counter.group()))
-                res++;
-        }
-
-        return res;
-    }
-
-    /**
-     * Returns counters iterator for specified group.
-     *
-     * @param grpName Name of the group to iterate.
-     * @return Counters iterator.
-     */
-    public Iterator<Counter> iterateGroup(String grpName) {
-        Collection<Counter> grpCounters = new ArrayList<>();
-
-        for (HadoopLongCounter counter : cntrs.values()) {
-            if (grpName.equals(counter.group()))
-                grpCounters.add(new HadoopV2Counter(counter));
-        }
-
-        return grpCounters.iterator();
-    }
-
-    /**
-     * Find a counter in the group.
-     *
-     * @param grpName The name of the counter group.
-     * @param cntrName The name of the counter.
-     * @param create Create the counter if not found if true.
-     * @return The counter that was found or added or {@code null} if create is false.
-     */
-    public Counter findCounter(String grpName, String cntrName, boolean create) {
-        T2<String, String> key = new T2<>(grpName, cntrName);
-
-        HadoopLongCounter internalCntr = cntrs.get(key);
-
-        if (internalCntr == null & create) {
-            internalCntr = new HadoopLongCounter(grpName,cntrName);
-
-            cntrs.put(key, new HadoopLongCounter(grpName,cntrName));
-        }
-
-        return internalCntr == null ? null : new HadoopV2Counter(internalCntr);
-    }
-}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/ignite/blob/8032fc2c/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopProcessor.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopProcessor.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopProcessor.java
index b9c20c3..f0df1e9 100644
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopProcessor.java
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopProcessor.java
@@ -40,6 +40,9 @@ import java.util.concurrent.atomic.AtomicInteger;
  * Hadoop processor.
  */
 public class HadoopProcessor extends HadoopProcessorAdapter {
+    /** Class to probe for Hadoop libraries in Ignite classpath. */
+    private static final String HADOOP_PROBE_CLS = "org.apache.hadoop.conf.Configuration";
+
     /** Job ID counter. */
     private final AtomicInteger idCtr = new AtomicInteger();
 
@@ -164,7 +167,14 @@ public class HadoopProcessor extends HadoopProcessorAdapter {
 
     /** {@inheritDoc} */
     @Override public IgniteInternalFuture<?> submit(HadoopJobId jobId, HadoopJobInfo jobInfo) {
-        return hctx.jobTracker().submit(jobId, jobInfo);
+        ClassLoader oldLdr = HadoopCommonUtils.setContextClassLoader(getClass().getClassLoader());
+
+        try {
+            return hctx.jobTracker().submit(jobId, jobInfo);
+        }
+        finally {
+            HadoopCommonUtils.restoreContextClassLoader(oldLdr);
+        }
     }
 
     /** {@inheritDoc} */
@@ -203,6 +213,26 @@ public class HadoopProcessor extends HadoopProcessorAdapter {
             throw new IgniteCheckedException(ioe.getMessage(), ioe);
         }
 
+        // Check if Hadoop is in parent class loader classpath.
+        try {
+            Class cls = Class.forName(HADOOP_PROBE_CLS, false, getClass().getClassLoader());
+
+            try {
+                String path = cls.getProtectionDomain().getCodeSource().getLocation().toString();
+
+                U.warn(log, "Hadoop libraries are found in Ignite classpath, this could lead to class loading " +
+                    "errors (please remove all Hadoop libraries from Ignite classpath) [path=" + path + ']');
+            }
+            catch (Throwable ignore) {
+                U.warn(log, "Hadoop libraries are found in Ignite classpath, this could lead to class loading " +
+                    "errors (please remove all Hadoop libraries from Ignite classpath)");
+            }
+        }
+        catch (Throwable ignore) {
+            // All is fine.
+        }
+
+        // Try assembling Hadoop URLs.
         HadoopClassLoader.hadoopUrls();
     }
 


Mime
View raw message