You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vo...@apache.org on 2015/03/03 14:08:21 UTC
[05/31] incubator-ignite git commit: # IGNITE-386: WIP on internal
namings.
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ace354c6/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
new file mode 100644
index 0000000..6915d17
--- /dev/null
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopClassLoader.java
@@ -0,0 +1,552 @@
+/*
+ * 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.*;
+import org.apache.ignite.internal.processors.hadoop.v2.*;
+import org.apache.ignite.internal.util.typedef.*;
+import org.jdk8.backport.*;
+import org.jetbrains.annotations.*;
+import org.objectweb.asm.*;
+import org.objectweb.asm.commons.*;
+
+import java.io.*;
+import java.net.*;
+import java.util.*;
+import java.util.concurrent.atomic.*;
+
+/**
+ * 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 {
+ /**
+ * We are very parallel capable.
+ */
+ static {
+ registerAsParallelCapable();
+ }
+
+ /** */
+ 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<>();
+
+ /**
+ * @param urls Urls.
+ */
+ public HadoopClassLoader(URL[] urls) {
+ super(addHadoopUrls(urls), APP_CLS_LDR);
+
+ assert !(getParent() instanceof HadoopClassLoader);
+ }
+
+ /**
+ * 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 isIgfsHadoop(String cls) {
+ String ignitePackagePrefix = "org.apache.ignite";
+ int len = ignitePackagePrefix.length();
+
+ return cls.startsWith(ignitePackagePrefix) && (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, GridHadoopShutdownHookManager.class.getName());
+ else if (name.endsWith(".util.NativeCodeLoader"))
+ return loadFromBytes(name, GridHadoopNativeCodeLoader.class.getName());
+
+ return loadClassExplicitly(name, resolve);
+ }
+
+ if (isIgfsHadoop(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, new HashSet<String>());
+
+ 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);
+ }
+ }
+
+ /**
+ * @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");
+ }
+
+ /**
+ * @param clsName Class name.
+ * @return {@code true} If the class has external dependencies.
+ */
+ boolean hasExternalDependencies(final String clsName, final Set<String> visited) {
+ 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 (!isIgfsHadoop(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);
+ }
+
+ visited.add(clsName);
+
+ final AtomicBoolean hasDeps = new AtomicBoolean();
+
+ rdr.accept(new ClassVisitor(Opcodes.ASM4) {
+ AnnotationVisitor av = new AnnotationVisitor(Opcodes.ASM4) {
+ // TODO
+ };
+
+ FieldVisitor fv = new FieldVisitor(Opcodes.ASM4) {
+ @Override public AnnotationVisitor visitAnnotation(String desc, boolean b) {
+ onType(desc);
+
+ return av;
+ }
+ };
+
+ MethodVisitor mv = new MethodVisitor(Opcodes.ASM4) {
+ @Override public AnnotationVisitor visitAnnotation(String desc, boolean b) {
+ onType(desc);
+
+ return av;
+ }
+
+ @Override public AnnotationVisitor visitParameterAnnotation(int i, String desc, boolean b) {
+ onType(desc);
+
+ return av;
+ }
+
+ @Override public AnnotationVisitor visitAnnotationDefault() {
+ return av;
+ }
+
+ @Override public void visitFieldInsn(int i, String owner, String name, String desc) {
+ onType(owner);
+ onType(desc);
+ }
+
+ @Override public void visitFrame(int i, int i2, Object[] locTypes, int i3, Object[] stackTypes) {
+ for (Object o : locTypes) {
+ if (o instanceof String)
+ onType((String)o);
+ }
+
+ for (Object o : stackTypes) {
+ if (o instanceof String)
+ onType((String)o);
+ }
+ }
+
+ @Override public void visitLocalVariable(String name, String desc, String signature, Label lb,
+ Label lb2, int i) {
+ onType(desc);
+ }
+
+ @Override public void visitMethodInsn(int i, String owner, String name, String desc) {
+ onType(owner);
+ }
+
+ @Override public void visitMultiANewArrayInsn(String desc, int dim) {
+ onType(desc);
+ }
+
+ @Override public void visitTryCatchBlock(Label lb, Label lb2, Label lb3, String e) {
+ onType(e);
+ }
+ };
+
+ void onClass(String depCls) {
+ assert validateClassName(depCls) : depCls;
+
+ if (depCls.startsWith("java.")) // Filter out platform classes.
+ return;
+
+ if (visited.contains(depCls))
+ return;
+
+ Boolean res = cache.get(depCls);
+
+ if (res == Boolean.TRUE || (res == null && hasExternalDependencies(depCls, visited)))
+ hasDeps.set(true);
+ }
+
+ void onType(String type) {
+ if (type == null)
+ return;
+
+ int off = 0;
+
+ while (type.charAt(off) == '[')
+ off++; // Handle arrays.
+
+ if (off != 0)
+ type = type.substring(off);
+
+ if (type.length() == 1)
+ return; // Get rid of primitives.
+
+ if (type.charAt(type.length() - 1) == ';') {
+ assert type.charAt(0) == 'L' : type;
+
+ type = type.substring(1, type.length() - 1);
+ }
+
+ type = type.replace('/', '.');
+
+ onClass(type);
+ }
+
+ @Override public void visit(int i, int i2, String name, String signature, String superName,
+ String[] ifaces) {
+ onType(superName);
+
+ if (ifaces != null) {
+ for (String iface : ifaces)
+ onType(iface);
+ }
+ }
+
+ @Override public AnnotationVisitor visitAnnotation(String desc, boolean visible) {
+ onType(desc);
+
+ return av;
+ }
+
+ @Override public void visitInnerClass(String name, String outerName, String innerName, int i) {
+ onType(name);
+ }
+
+ @Override public FieldVisitor visitField(int i, String name, String desc, String signature, Object val) {
+ onType(desc);
+
+ return fv;
+ }
+
+ @Override public MethodVisitor visitMethod(int i, String name, String desc, String signature,
+ String[] exceptions) {
+ if (exceptions != null) {
+ for (String e : exceptions)
+ onType(e);
+ }
+
+ return mv;
+ }
+ }, 0);
+
+ if (hasDeps.get()) // 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 have them.
+ int idx = clsName.lastIndexOf('$');
+
+ if (idx == -1) // No parent class.
+ return false;
+
+ String parentCls = clsName.substring(0, idx);
+
+ if (visited.contains(parentCls))
+ return false;
+
+ Boolean res = cache.get(parentCls);
+
+ if (res == null)
+ res = hasExternalDependencies(parentCls, visited);
+
+ 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 name Variable name.
+ * @param dflt Default.
+ * @return Value.
+ */
+ private static String getEnv(String name, String dflt) {
+ String res = System.getProperty(name);
+
+ if (F.isEmpty(res))
+ res = System.getenv(name);
+
+ return F.isEmpty(res) ? dflt : res;
+ }
+
+ /**
+ * @param res Result.
+ * @param dir Directory.
+ * @param startsWith Starts with prefix.
+ * @throws MalformedURLException If failed.
+ */
+ private static void addUrls(Collection<URL> res, File dir, final String startsWith) throws Exception {
+ File[] files = dir.listFiles(new FilenameFilter() {
+ @Override public boolean accept(File dir, String name) {
+ return startsWith == null || name.startsWith(startsWith);
+ }
+ });
+
+ if (files == null)
+ throw new IOException("Path is not a directory: " + dir);
+
+ for (File file : files)
+ res.add(file.toURI().toURL());
+ }
+
+ /**
+ * @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 HADOOP_HOME Variable.
+ */
+ @Nullable public static String hadoopHome() {
+ return getEnv("HADOOP_PREFIX", getEnv("HADOOP_HOME", null));
+ }
+
+ /**
+ * @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;
+
+ hadoopUrls = new ArrayList<>();
+
+ String hadoopPrefix = hadoopHome();
+
+ if (F.isEmpty(hadoopPrefix))
+ throw new IgniteCheckedException("Failed resolve Hadoop installation location. Either HADOOP_PREFIX or " +
+ "HADOOP_HOME environment variables must be set.");
+
+ String commonHome = getEnv("HADOOP_COMMON_HOME", hadoopPrefix + "/share/hadoop/common");
+ String hdfsHome = getEnv("HADOOP_HDFS_HOME", hadoopPrefix + "/share/hadoop/hdfs");
+ String mapredHome = getEnv("HADOOP_MAPRED_HOME", hadoopPrefix + "/share/hadoop/mapreduce");
+
+ try {
+ addUrls(hadoopUrls, new File(commonHome + "/lib"), null);
+ addUrls(hadoopUrls, new File(hdfsHome + "/lib"), null);
+ addUrls(hadoopUrls, new File(mapredHome + "/lib"), null);
+
+ addUrls(hadoopUrls, new File(hdfsHome), "hadoop-hdfs-");
+
+ addUrls(hadoopUrls, new File(commonHome), "hadoop-common-");
+ addUrls(hadoopUrls, new File(commonHome), "hadoop-auth-");
+ addUrls(hadoopUrls, new File(commonHome + "/lib"), "hadoop-auth-");
+
+ addUrls(hadoopUrls, new File(mapredHome), "hadoop-mapreduce-client-common");
+ addUrls(hadoopUrls, new File(mapredHome), "hadoop-mapreduce-client-core");
+ }
+ catch (Exception e) {
+ throw new IgniteCheckedException(e);
+ }
+
+ hadoopJars = hadoopUrls;
+
+ return hadoopUrls;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ace354c6/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopComponent.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopComponent.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopComponent.java
new file mode 100644
index 0000000..cea11eb
--- /dev/null
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopComponent.java
@@ -0,0 +1,61 @@
+/*
+ * 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.*;
+
+/**
+ * Abstract class for all hadoop components.
+ */
+public abstract class HadoopComponent {
+ /** Hadoop context. */
+ protected HadoopContext ctx;
+
+ /** Logger. */
+ protected IgniteLogger log;
+
+ /**
+ * @param ctx Hadoop context.
+ */
+ public void start(HadoopContext ctx) throws IgniteCheckedException {
+ this.ctx = ctx;
+
+ log = ctx.kernalContext().log(getClass());
+ }
+
+ /**
+ * Stops manager.
+ */
+ public void stop(boolean cancel) {
+ // No-op.
+ }
+
+ /**
+ * Callback invoked when all grid components are started.
+ */
+ public void onKernalStart() throws IgniteCheckedException {
+ // No-op.
+ }
+
+ /**
+ * Callback invoked before all grid components are stopped.
+ */
+ public void onKernalStop(boolean cancel) {
+ // No-op.
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ace354c6/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
new file mode 100644
index 0000000..bb707c8
--- /dev/null
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopContext.java
@@ -0,0 +1,196 @@
+/*
+ * 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.cluster.*;
+import org.apache.ignite.internal.*;
+import org.apache.ignite.internal.processors.hadoop.jobtracker.*;
+import org.apache.ignite.internal.processors.hadoop.shuffle.*;
+import org.apache.ignite.internal.processors.hadoop.taskexecutor.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+
+import java.util.*;
+
+/**
+ * Hadoop accelerator context.
+ */
+public class HadoopContext {
+ /** Kernal context. */
+ private GridKernalContext ctx;
+
+ /** Hadoop configuration. */
+ private GridHadoopConfiguration cfg;
+
+ /** Job tracker. */
+ private HadoopJobTracker jobTracker;
+
+ /** External task executor. */
+ private HadoopTaskExecutorAdapter taskExecutor;
+
+ /** */
+ private HadoopShuffle shuffle;
+
+ /** Managers list. */
+ private List<HadoopComponent> components = new ArrayList<>();
+
+ /**
+ * @param ctx Kernal context.
+ */
+ public HadoopContext(
+ GridKernalContext ctx,
+ GridHadoopConfiguration cfg,
+ HadoopJobTracker jobTracker,
+ HadoopTaskExecutorAdapter taskExecutor,
+ HadoopShuffle shuffle
+ ) {
+ this.ctx = ctx;
+ this.cfg = cfg;
+
+ this.jobTracker = add(jobTracker);
+ this.taskExecutor = add(taskExecutor);
+ this.shuffle = add(shuffle);
+ }
+
+ /**
+ * Gets list of managers.
+ *
+ * @return List of managers.
+ */
+ public List<HadoopComponent> components() {
+ return components;
+ }
+
+ /**
+ * Gets kernal context.
+ *
+ * @return Grid kernal context instance.
+ */
+ public GridKernalContext kernalContext() {
+ return ctx;
+ }
+
+ /**
+ * Gets Hadoop configuration.
+ *
+ * @return Hadoop configuration.
+ */
+ public GridHadoopConfiguration configuration() {
+ return cfg;
+ }
+
+ /**
+ * Gets local node ID. Shortcut for {@code kernalContext().localNodeId()}.
+ *
+ * @return Local node ID.
+ */
+ public UUID localNodeId() {
+ return ctx.localNodeId();
+ }
+
+ /**
+ * Gets local node order.
+ *
+ * @return Local node order.
+ */
+ public long localNodeOrder() {
+ assert ctx.discovery() != null;
+
+ return ctx.discovery().localNode().order();
+ }
+
+ /**
+ * @return Hadoop-enabled nodes.
+ */
+ public Collection<ClusterNode> nodes() {
+ return ctx.discovery().cacheNodes(CU.SYS_CACHE_HADOOP_MR, ctx.discovery().topologyVersion());
+ }
+
+ /**
+ * @return {@code True} if
+ */
+ public boolean jobUpdateLeader() {
+ long minOrder = Long.MAX_VALUE;
+ ClusterNode minOrderNode = null;
+
+ for (ClusterNode node : nodes()) {
+ if (node.order() < minOrder) {
+ minOrder = node.order();
+ minOrderNode = node;
+ }
+ }
+
+ assert minOrderNode != null;
+
+ return localNodeId().equals(minOrderNode.id());
+ }
+
+ /**
+ * @param meta Job metadata.
+ * @return {@code true} If local node is participating in job execution.
+ */
+ public boolean isParticipating(GridHadoopJobMetadata meta) {
+ UUID locNodeId = localNodeId();
+
+ if (locNodeId.equals(meta.submitNodeId()))
+ return true;
+
+ GridHadoopMapReducePlan plan = meta.mapReducePlan();
+
+ return plan.mapperNodeIds().contains(locNodeId) || plan.reducerNodeIds().contains(locNodeId) || jobUpdateLeader();
+ }
+
+ /**
+ * @return Jon tracker instance.
+ */
+ public HadoopJobTracker jobTracker() {
+ return jobTracker;
+ }
+
+ /**
+ * @return Task executor.
+ */
+ public HadoopTaskExecutorAdapter taskExecutor() {
+ return taskExecutor;
+ }
+
+ /**
+ * @return Shuffle.
+ */
+ public HadoopShuffle shuffle() {
+ return shuffle;
+ }
+
+ /**
+ * @return Map-reduce planner.
+ */
+ public GridHadoopMapReducePlanner planner() {
+ return cfg.getMapReducePlanner();
+ }
+
+ /**
+ * Adds component.
+ *
+ * @param c Component to add.
+ * @return Added manager.
+ */
+ private <C extends HadoopComponent> C add(C c) {
+ components.add(c);
+
+ return c;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ace354c6/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopCounterGroup.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopCounterGroup.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopCounterGroup.java
index bdf8fc6..8655e14 100644
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopCounterGroup.java
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopCounterGroup.java
@@ -24,7 +24,7 @@ import java.io.*;
import java.util.*;
/**
- * Hadoop counter group adapter.
+ * Hadoop +counter group adapter.
*/
class HadoopCounterGroup implements CounterGroup {
/** Counters. */
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ace354c6/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
new file mode 100644
index 0000000..370b82d
--- /dev/null
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopDefaultJobInfo.java
@@ -0,0 +1,163 @@
+/*
+ * 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.*;
+import org.apache.ignite.internal.processors.hadoop.v2.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+import org.jetbrains.annotations.*;
+
+import java.io.*;
+import java.lang.reflect.*;
+import java.util.*;
+
+/**
+ * Hadoop job info based on default Hadoop configuration.
+ */
+public class HadoopDefaultJobInfo implements GridHadoopJobInfo, 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;
+
+ /** */
+ private static volatile Class<?> jobCls;
+
+ /**
+ * 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 GridHadoopJob createJob(GridHadoopJobId jobId, IgniteLogger log) throws IgniteCheckedException {
+ try {
+ Class<?> jobCls0 = jobCls;
+
+ if (jobCls0 == null) { // It is enough to have only one class loader with only Hadoop classes.
+ synchronized (HadoopDefaultJobInfo.class) {
+ if ((jobCls0 = jobCls) == null) {
+ HadoopClassLoader ldr = new HadoopClassLoader(null);
+
+ jobCls = jobCls0 = ldr.loadClass(GridHadoopV2Job.class.getName());
+ }
+ }
+ }
+
+ Constructor<?> constructor = jobCls0.getConstructor(GridHadoopJobId.class, HadoopDefaultJobInfo.class,
+ IgniteLogger.class);
+
+ return (GridHadoopJob)constructor.newInstance(jobId, this, log);
+ }
+ // NB: java.lang.NoClassDefFoundError may be thrown from Class#getConstructor() call.
+ catch (Throwable 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;
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ace354c6/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopImpl.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopImpl.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopImpl.java
new file mode 100644
index 0000000..80fd995
--- /dev/null
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopImpl.java
@@ -0,0 +1,132 @@
+/*
+ * 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.*;
+import org.apache.ignite.internal.*;
+import org.apache.ignite.internal.util.*;
+import org.jetbrains.annotations.*;
+
+/**
+ * Hadoop facade implementation.
+ */
+public class HadoopImpl implements GridHadoop {
+ /** Hadoop processor. */
+ private final IgniteHadoopProcessor proc;
+
+ /** Busy lock. */
+ private final GridSpinBusyLock busyLock = new GridSpinBusyLock();
+
+ /**
+ * Constructor.
+ *
+ * @param proc Hadoop processor.
+ */
+ HadoopImpl(IgniteHadoopProcessor proc) {
+ this.proc = proc;
+ }
+
+ /** {@inheritDoc} */
+ @Override public GridHadoopConfiguration configuration() {
+ return proc.config();
+ }
+
+ /** {@inheritDoc} */
+ @Override public GridHadoopJobId nextJobId() {
+ if (busyLock.enterBusy()) {
+ try {
+ return proc.nextJobId();
+ }
+ finally {
+ busyLock.leaveBusy();
+ }
+ }
+ else
+ throw new IllegalStateException("Failed to get next job ID (grid is stopping).");
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteInternalFuture<?> submit(GridHadoopJobId jobId, GridHadoopJobInfo jobInfo) {
+ if (busyLock.enterBusy()) {
+ try {
+ return proc.submit(jobId, jobInfo);
+ }
+ finally {
+ busyLock.leaveBusy();
+ }
+ }
+ else
+ throw new IllegalStateException("Failed to submit job (grid is stopping).");
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public GridHadoopJobStatus status(GridHadoopJobId jobId) throws IgniteCheckedException {
+ if (busyLock.enterBusy()) {
+ try {
+ return proc.status(jobId);
+ }
+ finally {
+ busyLock.leaveBusy();
+ }
+ }
+ else
+ throw new IllegalStateException("Failed to get job status (grid is stopping).");
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public GridHadoopCounters counters(GridHadoopJobId jobId) throws IgniteCheckedException {
+ if (busyLock.enterBusy()) {
+ try {
+ return proc.counters(jobId);
+ }
+ finally {
+ busyLock.leaveBusy();
+ }
+ }
+ else
+ throw new IllegalStateException("Failed to get job counters (grid is stopping).");
+ }
+
+ /** {@inheritDoc} */
+ @Nullable @Override public IgniteInternalFuture<?> finishFuture(GridHadoopJobId jobId) throws IgniteCheckedException {
+ if (busyLock.enterBusy()) {
+ try {
+ return proc.finishFuture(jobId);
+ }
+ finally {
+ busyLock.leaveBusy();
+ }
+ }
+ else
+ throw new IllegalStateException("Failed to get job finish future (grid is stopping).");
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean kill(GridHadoopJobId jobId) throws IgniteCheckedException {
+ if (busyLock.enterBusy()) {
+ try {
+ return proc.kill(jobId);
+ }
+ finally {
+ busyLock.leaveBusy();
+ }
+ }
+ else
+ throw new IllegalStateException("Failed to kill job (grid is stopping).");
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ace354c6/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopSetup.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopSetup.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopSetup.java
new file mode 100644
index 0000000..35df5da
--- /dev/null
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopSetup.java
@@ -0,0 +1,505 @@
+/*
+ * 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.util.typedef.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+
+import java.io.*;
+import java.net.*;
+import java.nio.file.*;
+import java.text.*;
+import java.util.*;
+
+import static org.apache.ignite.internal.IgniteVersionUtils.*;
+
+/**
+ * Setup tool to configure Hadoop client.
+ */
+public class HadoopSetup {
+ /** */
+ public static final String WINUTILS_EXE = "winutils.exe";
+
+ /** */
+ private static final FilenameFilter IGNITE_JARS = new FilenameFilter() {
+ @Override public boolean accept(File dir, String name) {
+ return name.startsWith("ignite-") && name.endsWith(".jar");
+ }
+ };
+
+ /**
+ * The main method.
+ * @param ignore Params.
+ */
+ public static void main(String[] ignore) {
+ X.println(
+ " __________ ________________ ",
+ " / _/ ___/ |/ / _/_ __/ __/ ",
+ " _/ // (_ / // / / / / _/ ",
+ "/___/\\___/_/|_/___/ /_/ /___/ ",
+ " for Apache Hadoop ",
+ " ");
+
+ println("Version " + ACK_VER_STR);
+
+ configureHadoop();
+ }
+
+ /**
+ * This operation prepares the clean unpacked Hadoop distributive to work as client with Ignite-Hadoop.
+ * It performs these operations:
+ * <ul>
+ * <li>Check for setting of HADOOP_HOME environment variable.</li>
+ * <li>Try to resolve HADOOP_COMMON_HOME or evaluate it relative to HADOOP_HOME.</li>
+ * <li>In Windows check if winutils.exe exists and try to fix issue with some restrictions.</li>
+ * <li>In Windows check new line character issues in CMD scripts.</li>
+ * <li>Scan Hadoop lib directory to detect Ignite JARs. If these don't exist tries to create ones.</li>
+ * </ul>
+ */
+ private static void configureHadoop() {
+ String igniteHome = U.getIgniteHome();
+
+ println("IGNITE_HOME is set to '" + igniteHome + "'.");
+
+ checkIgniteHome(igniteHome);
+
+ String homeVar = "HADOOP_HOME";
+ String hadoopHome = System.getenv(homeVar);
+
+ if (F.isEmpty(hadoopHome)) {
+ homeVar = "HADOOP_PREFIX";
+ hadoopHome = System.getenv(homeVar);
+ }
+
+ if (F.isEmpty(hadoopHome))
+ exit("Neither HADOOP_HOME nor HADOOP_PREFIX environment variable is set. Please set one of them to a " +
+ "valid Hadoop installation directory and run setup tool again.", null);
+
+ hadoopHome = hadoopHome.replaceAll("\"", "");
+
+ println(homeVar + " is set to '" + hadoopHome + "'.");
+
+ String hiveHome = System.getenv("HIVE_HOME");
+
+ if (!F.isEmpty(hiveHome)) {
+ hiveHome = hiveHome.replaceAll("\"", "");
+
+ println("HIVE_HOME is set to '" + hiveHome + "'.");
+ }
+
+ File hadoopDir = new File(hadoopHome);
+
+ if (!hadoopDir.exists())
+ exit("Hadoop installation folder does not exist.", null);
+
+ if (!hadoopDir.isDirectory())
+ exit("HADOOP_HOME must point to a directory.", null);
+
+ if (!hadoopDir.canRead())
+ exit("Hadoop installation folder can not be read. Please check permissions.", null);
+
+ File hadoopCommonDir;
+
+ String hadoopCommonHome = System.getenv("HADOOP_COMMON_HOME");
+
+ if (F.isEmpty(hadoopCommonHome)) {
+ hadoopCommonDir = new File(hadoopDir, "share/hadoop/common");
+
+ println("HADOOP_COMMON_HOME is not set, will use '" + hadoopCommonDir.getPath() + "'.");
+ }
+ else {
+ println("HADOOP_COMMON_HOME is set to '" + hadoopCommonHome + "'.");
+
+ hadoopCommonDir = new File(hadoopCommonHome);
+ }
+
+ if (!hadoopCommonDir.canRead())
+ exit("Failed to read Hadoop common dir in '" + hadoopCommonHome + "'.", null);
+
+ File hadoopCommonLibDir = new File(hadoopCommonDir, "lib");
+
+ if (!hadoopCommonLibDir.canRead())
+ exit("Failed to read Hadoop 'lib' folder in '" + hadoopCommonLibDir.getPath() + "'.", null);
+
+ if (U.isWindows()) {
+ checkJavaPathSpaces();
+
+ File hadoopBinDir = new File(hadoopDir, "bin");
+
+ if (!hadoopBinDir.canRead())
+ exit("Failed to read subdirectory 'bin' in HADOOP_HOME.", null);
+
+ File winutilsFile = new File(hadoopBinDir, WINUTILS_EXE);
+
+ if (!winutilsFile.exists()) {
+ if (ask("File '" + WINUTILS_EXE + "' does not exist. " +
+ "It may be replaced by a stub. Create it?")) {
+ println("Creating file stub '" + winutilsFile.getAbsolutePath() + "'.");
+
+ boolean ok = false;
+
+ try {
+ ok = winutilsFile.createNewFile();
+ }
+ catch (IOException ignore) {
+ // No-op.
+ }
+
+ if (!ok)
+ exit("Failed to create '" + WINUTILS_EXE + "' file. Please check permissions.", null);
+ }
+ else
+ println("Ok. But Hadoop client probably will not work on Windows this way...");
+ }
+
+ processCmdFiles(hadoopDir, "bin", "sbin", "libexec");
+ }
+
+ File igniteLibs = new File(new File(igniteHome), "libs");
+
+ if (!igniteLibs.exists())
+ exit("Ignite 'libs' folder is not found.", null);
+
+ Collection<File> jarFiles = new ArrayList<>();
+
+ addJarsInFolder(jarFiles, igniteLibs);
+ addJarsInFolder(jarFiles, new File(igniteLibs, "ignite-hadoop"));
+
+ boolean jarsLinksCorrect = true;
+
+ for (File file : jarFiles) {
+ File link = new File(hadoopCommonLibDir, file.getName());
+
+ jarsLinksCorrect &= isJarLinkCorrect(link, file);
+
+ if (!jarsLinksCorrect)
+ break;
+ }
+
+ if (!jarsLinksCorrect) {
+ if (ask("Ignite JAR files are not found in Hadoop 'lib' directory. " +
+ "Create appropriate symbolic links?")) {
+ File[] oldIgniteJarFiles = hadoopCommonLibDir.listFiles(IGNITE_JARS);
+
+ if (oldIgniteJarFiles.length > 0 && ask("The Hadoop 'lib' directory contains JARs from other Ignite " +
+ "installation. They must be deleted to continue. Continue?")) {
+ for (File file : oldIgniteJarFiles) {
+ println("Deleting file '" + file.getAbsolutePath() + "'.");
+
+ if (!file.delete())
+ exit("Failed to delete file '" + file.getPath() + "'.", null);
+ }
+ }
+
+ for (File file : jarFiles) {
+ File targetFile = new File(hadoopCommonLibDir, file.getName());
+
+ try {
+ println("Creating symbolic link '" + targetFile.getAbsolutePath() + "'.");
+
+ Files.createSymbolicLink(targetFile.toPath(), file.toPath());
+ }
+ catch (IOException e) {
+ if (U.isWindows()) {
+ warn("Ability to create symbolic links is required!");
+ warn("On Windows platform you have to grant permission 'Create symbolic links'");
+ warn("to your user or run the Accelerator as Administrator.");
+ }
+
+ exit("Creating symbolic link failed! Check permissions.", e);
+ }
+ }
+ }
+ else
+ println("Ok. But Hadoop client will not be able to talk to Ignite cluster without those JARs in classpath...");
+ }
+
+ File hadoopEtc = new File(hadoopDir, "etc" + File.separator + "hadoop");
+
+ File igniteDocs = new File(igniteHome, "docs");
+
+ if (!igniteDocs.canRead())
+ exit("Failed to read Ignite 'docs' folder at '" + igniteDocs.getAbsolutePath() + "'.", null);
+
+ if (hadoopEtc.canWrite()) { // TODO Bigtop
+ if (ask("Replace 'core-site.xml' and 'mapred-site.xml' files with preconfigured templates " +
+ "(existing files will be backed up)?")) {
+ replaceWithBackup(new File(igniteDocs, "core-site.ignite.xml"), new File(hadoopEtc, "core-site.xml"));
+
+ replaceWithBackup(new File(igniteDocs, "mapred-site.ignite.xml"), new File(hadoopEtc, "mapred-site.xml"));
+ }
+ else
+ println("Ok. You can configure them later, the templates are available at Ignite's 'docs' directory...");
+ }
+
+ if (!F.isEmpty(hiveHome)) {
+ File hiveConfDir = new File(hiveHome + File.separator + "conf");
+
+ if (!hiveConfDir.canWrite())
+ warn("Can not write to '" + hiveConfDir.getAbsolutePath() + "'. To run Hive queries you have to " +
+ "configure 'hive-site.xml' manually. The template is available at Ignite's 'docs' directory.");
+ else if (ask("Replace 'hive-site.xml' with preconfigured template (existing file will be backed up)?"))
+ replaceWithBackup(new File(igniteDocs, "hive-site.ignite.xml"), new File(hiveConfDir, "hive-site.xml"));
+ else
+ println("Ok. You can configure it later, the template is available at Ignite's 'docs' directory...");
+ }
+
+ println("Apache Hadoop setup is complete.");
+ }
+
+ /**
+ * @param jarFiles Jars.
+ * @param folder Folder.
+ */
+ private static void addJarsInFolder(Collection<File> jarFiles, File folder) {
+ if (!folder.exists())
+ exit("Folder '" + folder.getAbsolutePath() + "' is not found.", null);
+
+ jarFiles.addAll(Arrays.asList(folder.listFiles(IGNITE_JARS)));
+ }
+
+ /**
+ * Checks that JAVA_HOME does not contain space characters.
+ */
+ private static void checkJavaPathSpaces() {
+ String javaHome = System.getProperty("java.home");
+
+ if (javaHome.contains(" ")) {
+ warn("Java installation path contains space characters!");
+ warn("Hadoop client will not be able to start using '" + javaHome + "'.");
+ warn("Please install JRE to path which does not contain spaces and point JAVA_HOME to that installation.");
+ }
+ }
+
+ /**
+ * Checks Ignite home.
+ *
+ * @param igniteHome Ignite home.
+ */
+ private static void checkIgniteHome(String igniteHome) {
+ URL jarUrl = U.class.getProtectionDomain().getCodeSource().getLocation();
+
+ try {
+ Path jar = Paths.get(jarUrl.toURI());
+ Path igHome = Paths.get(igniteHome);
+
+ if (!jar.startsWith(igHome))
+ exit("Ignite JAR files are not under IGNITE_HOME.", null);
+ }
+ catch (Exception e) {
+ exit(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * Replaces target file with source file.
+ *
+ * @param from From.
+ * @param to To.
+ */
+ private static void replaceWithBackup(File from, File to) {
+ if (!from.canRead())
+ exit("Failed to read source file '" + from.getAbsolutePath() + "'.", null);
+
+ println("Replacing file '" + to.getAbsolutePath() + "'.");
+
+ try {
+ U.copy(from, renameToBak(to), true);
+ }
+ catch (IOException e) {
+ exit("Failed to replace file '" + to.getAbsolutePath() + "'.", e);
+ }
+ }
+
+ /**
+ * Renames file for backup.
+ *
+ * @param file File.
+ * @return File.
+ */
+ private static File renameToBak(File file) {
+ DateFormat fmt = new SimpleDateFormat("yyyy-MM-dd_HH-mm-ss");
+
+ if (file.exists() && !file.renameTo(new File(file.getAbsolutePath() + "." + fmt.format(new Date()) + ".bak")))
+ exit("Failed to rename file '" + file.getPath() + "'.", null);
+
+ return file;
+ }
+
+ /**
+ * Checks if link is correct.
+ *
+ * @param link Symbolic link.
+ * @param correctTarget Correct link target.
+ * @return {@code true} If link target is correct.
+ */
+ private static boolean isJarLinkCorrect(File link, File correctTarget) {
+ if (!Files.isSymbolicLink(link.toPath()))
+ return false; // It is a real file or it does not exist.
+
+ Path target = null;
+
+ try {
+ target = Files.readSymbolicLink(link.toPath());
+ }
+ catch (IOException e) {
+ exit("Failed to read symbolic link: " + link.getAbsolutePath(), e);
+ }
+
+ return Files.exists(target) && target.toFile().equals(correctTarget);
+ }
+
+ /**
+ * Writes the question end read the boolean answer from the console.
+ *
+ * @param question Question to write.
+ * @return {@code true} if user inputs 'Y' or 'y', {@code false} otherwise.
+ */
+ private static boolean ask(String question) {
+ X.println();
+ X.print(" < " + question + " (Y/N): ");
+
+ String answer = null;
+
+ if (!F.isEmpty(System.getenv("IGNITE_HADOOP_SETUP_YES")))
+ answer = "Y";
+ else {
+ BufferedReader br = new BufferedReader(new InputStreamReader(System.in));
+
+ try {
+ answer = br.readLine();
+ }
+ catch (IOException e) {
+ exit("Failed to read answer: " + e.getMessage(), e);
+ }
+ }
+
+ if (answer != null && "Y".equals(answer.toUpperCase().trim())) {
+ X.println(" > Yes.");
+
+ return true;
+ }
+ else {
+ X.println(" > No.");
+
+ return false;
+ }
+ }
+
+ /**
+ * Exit with message.
+ *
+ * @param msg Exit message.
+ */
+ private static void exit(String msg, Exception e) {
+ X.println(" ");
+ X.println(" # " + msg);
+ X.println(" # Setup failed, exiting... ");
+
+ if (e != null && !F.isEmpty(System.getenv("IGNITE_HADOOP_SETUP_DEBUG")))
+ e.printStackTrace();
+
+ System.exit(1);
+ }
+
+ /**
+ * Prints message.
+ *
+ * @param msg Message.
+ */
+ private static void println(String msg) {
+ X.println(" > " + msg);
+ }
+
+ /**
+ * Prints warning.
+ *
+ * @param msg Message.
+ */
+ private static void warn(String msg) {
+ X.println(" ! " + msg);
+ }
+
+ /**
+ * Checks that CMD files have valid MS Windows new line characters. If not, writes question to console and reads the
+ * answer. If it's 'Y' then backups original files and corrects invalid new line characters.
+ *
+ * @param rootDir Root directory to process.
+ * @param dirs Directories inside of the root to process.
+ */
+ private static void processCmdFiles(File rootDir, String... dirs) {
+ boolean answer = false;
+
+ for (String dir : dirs) {
+ File subDir = new File(rootDir, dir);
+
+ File[] cmdFiles = subDir.listFiles(new FilenameFilter() {
+ @Override public boolean accept(File dir, String name) {
+ return name.toLowerCase().endsWith(".cmd");
+ }
+ });
+
+ for (File file : cmdFiles) {
+ String content = null;
+
+ try (Scanner scanner = new Scanner(file)) {
+ content = scanner.useDelimiter("\\Z").next();
+ }
+ catch (FileNotFoundException e) {
+ exit("Failed to read file '" + file + "'.", e);
+ }
+
+ boolean invalid = false;
+
+ for (int i = 0; i < content.length(); i++) {
+ if (content.charAt(i) == '\n' && (i == 0 || content.charAt(i - 1) != '\r')) {
+ invalid = true;
+
+ break;
+ }
+ }
+
+ if (invalid) {
+ answer = answer || ask("One or more *.CMD files has invalid new line character. Replace them?");
+
+ if (!answer) {
+ println("Ok. But Windows most probably will fail to execute them...");
+
+ return;
+ }
+
+ println("Fixing newline characters in file '" + file.getAbsolutePath() + "'.");
+
+ renameToBak(file);
+
+ try (BufferedWriter writer = new BufferedWriter(new FileWriter(file))) {
+ for (int i = 0; i < content.length(); i++) {
+ if (content.charAt(i) == '\n' && (i == 0 || content.charAt(i - 1) != '\r'))
+ writer.write("\r");
+
+ writer.write(content.charAt(i));
+ }
+ }
+ catch (IOException e) {
+ exit("Failed to write file '" + file.getPath() + "': " + e.getMessage(), e);
+ }
+ }
+ }
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ace354c6/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopTaskCancelledException.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopTaskCancelledException.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopTaskCancelledException.java
new file mode 100644
index 0000000..bb3d1cc
--- /dev/null
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopTaskCancelledException.java
@@ -0,0 +1,35 @@
+/*
+ * 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.*;
+
+/**
+ * Exception that throws when the task is cancelling.
+ */
+public class HadoopTaskCancelledException extends IgniteException {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /**
+ * @param msg Exception message.
+ */
+ public HadoopTaskCancelledException(String msg) {
+ super(msg);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ace354c6/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopUtils.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopUtils.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopUtils.java
new file mode 100644
index 0000000..46594ce
--- /dev/null
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/HadoopUtils.java
@@ -0,0 +1,308 @@
+/*
+ * 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.hadoop.conf.*;
+import org.apache.hadoop.fs.*;
+import org.apache.hadoop.io.*;
+import org.apache.hadoop.mapred.*;
+import org.apache.hadoop.mapreduce.JobID;
+import org.apache.hadoop.mapreduce.JobPriority;
+import org.apache.hadoop.mapreduce.JobStatus;
+import org.apache.hadoop.mapreduce.*;
+import org.apache.ignite.*;
+import org.apache.ignite.internal.processors.hadoop.v2.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+
+import java.io.*;
+import java.util.*;
+
+/**
+ * Hadoop utility methods.
+ */
+public class HadoopUtils {
+ /** 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";
+
+ /** Staging constant. */
+ private static final String STAGING_CONSTANT = ".staging";
+
+ /** Old mapper class attribute. */
+ private static final String OLD_MAP_CLASS_ATTR = "mapred.mapper.class";
+
+ /** Old reducer class attribute. */
+ private static final String OLD_REDUCE_CLASS_ATTR = "mapred.reducer.class";
+
+ /**
+ * Wraps native split.
+ *
+ * @param id Split ID.
+ * @param split Split.
+ * @param hosts Hosts.
+ * @throws IOException If failed.
+ */
+ public static GridHadoopSplitWrapper wrapSplit(int id, Object split, String[] hosts) throws IOException {
+ ByteArrayOutputStream arr = new ByteArrayOutputStream();
+ ObjectOutput out = new ObjectOutputStream(arr);
+
+ assert split instanceof Writable;
+
+ ((Writable)split).write(out);
+
+ out.flush();
+
+ return new GridHadoopSplitWrapper(id, split.getClass().getName(), arr.toByteArray(), hosts);
+ }
+
+ /**
+ * Unwraps native split.
+ *
+ * @param o Wrapper.
+ * @return Split.
+ */
+ public static Object unwrapSplit(GridHadoopSplitWrapper o) {
+ try {
+ Writable w = (Writable)HadoopUtils.class.getClassLoader().loadClass(o.className()).newInstance();
+
+ w.readFields(new ObjectInputStream(new ByteArrayInputStream(o.bytes())));
+
+ return w;
+ }
+ catch (Exception e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
+ /**
+ * Convert Ignite job status to Hadoop job status.
+ *
+ * @param status Ignite job status.
+ * @return Hadoop job status.
+ */
+ public static JobStatus status(GridHadoopJobStatus status, Configuration conf) {
+ JobID jobId = new JobID(status.jobId().globalId().toString(), status.jobId().localId());
+
+ float setupProgress = 0;
+ float mapProgress = 0;
+ float reduceProgress = 0;
+ float cleanupProgress = 0;
+
+ JobStatus.State state = JobStatus.State.RUNNING;
+
+ switch (status.jobPhase()) {
+ case PHASE_SETUP:
+ setupProgress = 0.42f;
+
+ break;
+
+ case PHASE_MAP:
+ setupProgress = 1;
+ mapProgress = 1f - status.pendingMapperCnt() / (float)status.totalMapperCnt();
+
+ break;
+
+ case PHASE_REDUCE:
+ assert status.totalReducerCnt() > 0;
+
+ setupProgress = 1;
+ mapProgress = 1;
+ reduceProgress = 1f - status.pendingReducerCnt() / (float)status.totalReducerCnt();
+
+ break;
+
+ case PHASE_CANCELLING:
+ case PHASE_COMPLETE:
+ if (!status.isFailed()) {
+ setupProgress = 1;
+ mapProgress = 1;
+ reduceProgress = 1;
+ cleanupProgress = 1;
+
+ state = JobStatus.State.SUCCEEDED;
+ }
+ else
+ state = JobStatus.State.FAILED;
+
+ break;
+
+ default:
+ assert false;
+ }
+
+ return new JobStatus(jobId, setupProgress, mapProgress, reduceProgress, cleanupProgress, state,
+ JobPriority.NORMAL, status.user(), status.jobName(), jobFile(conf, status.user(), jobId).toString(), "N/A");
+ }
+
+ /**
+ * Gets staging area directory.
+ *
+ * @param conf Configuration.
+ * @param usr User.
+ * @return Staging area directory.
+ */
+ public static Path stagingAreaDir(Configuration conf, String usr) {
+ return new Path(conf.get(MRJobConfig.MR_AM_STAGING_DIR, MRJobConfig.DEFAULT_MR_AM_STAGING_DIR)
+ + Path.SEPARATOR + usr + Path.SEPARATOR + STAGING_CONSTANT);
+ }
+
+ /**
+ * Gets job file.
+ *
+ * @param conf Configuration.
+ * @param usr User.
+ * @param jobId Job ID.
+ * @return Job file.
+ */
+ public static Path jobFile(Configuration conf, String usr, JobID jobId) {
+ return new Path(stagingAreaDir(conf, usr), jobId.toString() + Path.SEPARATOR + MRJobConfig.JOB_CONF_FILE);
+ }
+
+ /**
+ * Checks the attribute in configuration is not set.
+ *
+ * @param attr Attribute name.
+ * @param msg Message for creation of exception.
+ * @throws IgniteCheckedException If attribute is set.
+ */
+ public static void ensureNotSet(Configuration cfg, String attr, String msg) throws IgniteCheckedException {
+ if (cfg.get(attr) != null)
+ throw new IgniteCheckedException(attr + " is incompatible with " + msg + " mode.");
+ }
+
+ /**
+ * Creates JobInfo from hadoop configuration.
+ *
+ * @param cfg Hadoop configuration.
+ * @return Job info.
+ * @throws IgniteCheckedException If failed.
+ */
+ public static HadoopDefaultJobInfo createJobInfo(Configuration cfg) throws IgniteCheckedException {
+ JobConf jobConf = new JobConf(cfg);
+
+ boolean hasCombiner = jobConf.get("mapred.combiner.class") != null
+ || jobConf.get(MRJobConfig.COMBINE_CLASS_ATTR) != null;
+
+ int numReduces = jobConf.getNumReduceTasks();
+
+ jobConf.setBooleanIfUnset("mapred.mapper.new-api", jobConf.get(OLD_MAP_CLASS_ATTR) == null);
+
+ if (jobConf.getUseNewMapper()) {
+ String mode = "new map API";
+
+ ensureNotSet(jobConf, "mapred.input.format.class", mode);
+ ensureNotSet(jobConf, OLD_MAP_CLASS_ATTR, mode);
+
+ if (numReduces != 0)
+ ensureNotSet(jobConf, "mapred.partitioner.class", mode);
+ else
+ ensureNotSet(jobConf, "mapred.output.format.class", mode);
+ }
+ else {
+ String mode = "map compatibility";
+
+ ensureNotSet(jobConf, MRJobConfig.INPUT_FORMAT_CLASS_ATTR, mode);
+ ensureNotSet(jobConf, MRJobConfig.MAP_CLASS_ATTR, mode);
+
+ if (numReduces != 0)
+ ensureNotSet(jobConf, MRJobConfig.PARTITIONER_CLASS_ATTR, mode);
+ else
+ ensureNotSet(jobConf, MRJobConfig.OUTPUT_FORMAT_CLASS_ATTR, mode);
+ }
+
+ if (numReduces != 0) {
+ jobConf.setBooleanIfUnset("mapred.reducer.new-api", jobConf.get(OLD_REDUCE_CLASS_ATTR) == null);
+
+ if (jobConf.getUseNewReducer()) {
+ String mode = "new reduce API";
+
+ ensureNotSet(jobConf, "mapred.output.format.class", mode);
+ ensureNotSet(jobConf, OLD_REDUCE_CLASS_ATTR, mode);
+ }
+ else {
+ String mode = "reduce compatibility";
+
+ ensureNotSet(jobConf, MRJobConfig.OUTPUT_FORMAT_CLASS_ATTR, mode);
+ ensureNotSet(jobConf, MRJobConfig.REDUCE_CLASS_ATTR, mode);
+ }
+ }
+
+ Map<String, String> props = new HashMap<>();
+
+ for (Map.Entry<String, String> entry : jobConf)
+ props.put(entry.getKey(), entry.getValue());
+
+ return new HadoopDefaultJobInfo(jobConf.getJobName(), jobConf.getUser(), hasCombiner, numReduces, props);
+ }
+
+ /**
+ * Throws new {@link IgniteCheckedException} with original exception is serialized into string.
+ * This is needed to transfer error outside the current class loader.
+ *
+ * @param e Original exception.
+ * @return IgniteCheckedException New exception.
+ */
+ public static IgniteCheckedException transformException(Throwable e) {
+ ByteArrayOutputStream os = new ByteArrayOutputStream();
+
+ e.printStackTrace(new PrintStream(os, true));
+
+ return new IgniteCheckedException(os.toString());
+ }
+
+ /**
+ * Returns work directory for job execution.
+ *
+ * @param locNodeId Local node ID.
+ * @param jobId Job ID.
+ * @return Working directory for job.
+ * @throws IgniteCheckedException If Failed.
+ */
+ public static File jobLocalDir(UUID locNodeId, GridHadoopJobId jobId) throws IgniteCheckedException {
+ return new File(new File(U.resolveWorkDirectory("hadoop", false), "node-" + locNodeId), "job_" + jobId);
+ }
+
+ /**
+ * Returns subdirectory of job working directory for task execution.
+ *
+ * @param locNodeId Local node ID.
+ * @param info Task info.
+ * @return Working directory for task.
+ * @throws IgniteCheckedException If Failed.
+ */
+ public static File taskLocalDir(UUID locNodeId, GridHadoopTaskInfo info) throws IgniteCheckedException {
+ File jobLocDir = jobLocalDir(locNodeId, info.jobId());
+
+ return new File(jobLocDir, info.type() + "_" + info.taskNumber() + "_" + info.attempt());
+ }
+
+ /**
+ * Constructor.
+ */
+ private HadoopUtils() {
+ // No-op.
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ace354c6/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/IgniteHadoopProcessor.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/IgniteHadoopProcessor.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/IgniteHadoopProcessor.java
index 4ef9e35..63e4854 100644
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/IgniteHadoopProcessor.java
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/IgniteHadoopProcessor.java
@@ -30,7 +30,7 @@ import org.apache.ignite.internal.util.typedef.internal.*;
import java.util.*;
import java.util.concurrent.atomic.*;
-import static org.apache.ignite.internal.processors.hadoop.GridHadoopClassLoader.*;
+import static org.apache.ignite.internal.processors.hadoop.HadoopClassLoader.*;
/**
* Hadoop processor.
@@ -41,7 +41,7 @@ public class IgniteHadoopProcessor extends IgniteHadoopProcessorAdapter {
/** Hadoop context. */
@GridToStringExclude
- private GridHadoopContext hctx;
+ private HadoopContext hctx;
/** Hadoop facade for public API. */
@GridToStringExclude
@@ -85,18 +85,18 @@ public class IgniteHadoopProcessor extends IgniteHadoopProcessorAdapter {
}
if (ok) {
- hctx = new GridHadoopContext(
+ hctx = new HadoopContext(
ctx,
cfg,
- new GridHadoopJobTracker(),
- cfg.isExternalExecution() ? new GridHadoopExternalTaskExecutor() : new GridHadoopEmbeddedTaskExecutor(),
- new GridHadoopShuffle());
+ new HadoopJobTracker(),
+ cfg.isExternalExecution() ? new HadoopExternalTaskExecutor() : new HadoopEmbeddedTaskExecutor(),
+ new HadoopShuffle());
- for (GridHadoopComponent c : hctx.components())
+ for (HadoopComponent c : hctx.components())
c.start(hctx);
- hadoop = new GridHadoopImpl(this);
+ hadoop = new HadoopImpl(this);
}
}
@@ -112,10 +112,10 @@ public class IgniteHadoopProcessor extends IgniteHadoopProcessorAdapter {
if (hctx == null)
return;
- List<GridHadoopComponent> components = hctx.components();
+ List<HadoopComponent> components = hctx.components();
- for (ListIterator<GridHadoopComponent> it = components.listIterator(components.size()); it.hasPrevious();) {
- GridHadoopComponent c = it.previous();
+ for (ListIterator<HadoopComponent> it = components.listIterator(components.size()); it.hasPrevious();) {
+ HadoopComponent c = it.previous();
c.stop(cancel);
}
@@ -128,7 +128,7 @@ public class IgniteHadoopProcessor extends IgniteHadoopProcessorAdapter {
if (hctx == null)
return;
- for (GridHadoopComponent c : hctx.components())
+ for (HadoopComponent c : hctx.components())
c.onKernalStart();
}
@@ -139,10 +139,10 @@ public class IgniteHadoopProcessor extends IgniteHadoopProcessorAdapter {
if (hctx == null)
return;
- List<GridHadoopComponent> components = hctx.components();
+ List<HadoopComponent> components = hctx.components();
- for (ListIterator<GridHadoopComponent> it = components.listIterator(components.size()); it.hasPrevious();) {
- GridHadoopComponent c = it.previous();
+ for (ListIterator<HadoopComponent> it = components.listIterator(components.size()); it.hasPrevious();) {
+ HadoopComponent c = it.previous();
c.onKernalStop(cancel);
}
@@ -153,7 +153,7 @@ public class IgniteHadoopProcessor extends IgniteHadoopProcessorAdapter {
*
* @return Hadoop context.
*/
- public GridHadoopContext context() {
+ public HadoopContext context() {
return hctx;
}
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ace354c6/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/counter/GridHadoopFSCounterWriter.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/counter/GridHadoopFSCounterWriter.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/counter/GridHadoopFSCounterWriter.java
index 55dcc4c..d603d76 100644
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/counter/GridHadoopFSCounterWriter.java
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/counter/GridHadoopFSCounterWriter.java
@@ -52,7 +52,7 @@ public class GridHadoopFSCounterWriter implements GridHadoopCounterWriter {
Configuration hadoopCfg = new Configuration();
- for (Map.Entry<String, String> e : ((GridHadoopDefaultJobInfo)jobInfo).properties().entrySet())
+ for (Map.Entry<String, String> e : ((HadoopDefaultJobInfo)jobInfo).properties().entrySet())
hadoopCfg.set(e.getKey(), e.getValue());
String user = jobInfo.user();
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/ace354c6/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/counter/GridHadoopPerformanceCounter.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/counter/GridHadoopPerformanceCounter.java b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/counter/GridHadoopPerformanceCounter.java
index d5ceebf..263a075 100644
--- a/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/counter/GridHadoopPerformanceCounter.java
+++ b/modules/hadoop/src/main/java/org/apache/ignite/internal/processors/hadoop/counter/GridHadoopPerformanceCounter.java
@@ -25,7 +25,7 @@ import org.jetbrains.annotations.*;
import java.io.*;
import java.util.*;
-import static org.apache.ignite.internal.processors.hadoop.GridHadoopUtils.*;
+import static org.apache.ignite.internal.processors.hadoop.HadoopUtils.*;
/**
* Counter for the job statistics accumulation.