ignite-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From anovi...@apache.org
Subject [06/60] [abbrv] incubator-ignite git commit: IGNITE-386: Squashed changes.
Date Thu, 05 Mar 2015 14:46:33 GMT
http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/6423cf02/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopJobTrackerSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopJobTrackerSelfTest.java b/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopJobTrackerSelfTest.java
deleted file mode 100644
index 3aa74d0..0000000
--- a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopJobTrackerSelfTest.java
+++ /dev/null
@@ -1,330 +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 org.apache.hadoop.fs.*;
-import org.apache.hadoop.mapreduce.*;
-import org.apache.hadoop.mapreduce.lib.input.*;
-import org.apache.hadoop.mapreduce.lib.output.*;
-import org.apache.ignite.internal.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-
-import java.io.*;
-import java.net.*;
-import java.util.*;
-import java.util.concurrent.*;
-import java.util.concurrent.atomic.*;
-
-import static org.apache.ignite.internal.processors.hadoop.GridHadoopUtils.*;
-
-/**
- * Job tracker self test.
- */
-public class GridHadoopJobTrackerSelfTest extends GridHadoopAbstractSelfTest {
-    /** */
-    private static final String PATH_OUTPUT = "/test-out";
-
-    /** Test block count parameter name. */
-    private static final int BLOCK_CNT = 10;
-
-    /** */
-    private static GridHadoopSharedMap m = GridHadoopSharedMap.map(GridHadoopJobTrackerSelfTest.class);
-
-    /** Map task execution count. */
-    private static final AtomicInteger mapExecCnt = m.put("mapExecCnt", new AtomicInteger());
-
-    /** Reduce task execution count. */
-    private static final AtomicInteger reduceExecCnt = m.put("reduceExecCnt", new AtomicInteger());
-
-    /** Reduce task execution count. */
-    private static final AtomicInteger combineExecCnt = m.put("combineExecCnt", new AtomicInteger());
-
-    /** */
-    private static final Map<String, CountDownLatch> latch = m.put("latch", new HashMap<String, CountDownLatch>());
-
-    /** {@inheritDoc} */
-    @Override protected boolean igfsEnabled() {
-        return true;
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void beforeTestsStarted() throws Exception {
-        super.beforeTestsStarted();
-
-        startGrids(gridCount());
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void afterTestsStopped() throws Exception {
-        stopAllGrids();
-
-        super.afterTestsStopped();
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void beforeTest() throws Exception {
-        latch.put("mapAwaitLatch", new CountDownLatch(1));
-        latch.put("reduceAwaitLatch", new CountDownLatch(1));
-        latch.put("combineAwaitLatch", new CountDownLatch(1));
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void afterTest() throws Exception {
-        mapExecCnt.set(0);
-        combineExecCnt.set(0);
-        reduceExecCnt.set(0);
-    }
-
-    /** {@inheritDoc} */
-    @Override public GridHadoopConfiguration hadoopConfiguration(String gridName) {
-        GridHadoopConfiguration cfg = super.hadoopConfiguration(gridName);
-
-        cfg.setMapReducePlanner(new GridHadoopTestRoundRobinMrPlanner());
-        cfg.setExternalExecution(false);
-
-        return cfg;
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testSimpleTaskSubmit() throws Exception {
-        try {
-            UUID globalId = UUID.randomUUID();
-
-            Job job = Job.getInstance();
-            setupFileSystems(job.getConfiguration());
-
-            job.setMapperClass(TestMapper.class);
-            job.setReducerClass(TestReducer.class);
-            job.setInputFormatClass(InFormat.class);
-
-            FileOutputFormat.setOutputPath(job, new Path(igfsScheme() + PATH_OUTPUT + "1"));
-
-            GridHadoopJobId jobId = new GridHadoopJobId(globalId, 1);
-
-            grid(0).hadoop().submit(jobId, createJobInfo(job.getConfiguration()));
-
-            checkStatus(jobId, false);
-
-            info("Releasing map latch.");
-
-            latch.get("mapAwaitLatch").countDown();
-
-            checkStatus(jobId, false);
-
-            info("Releasing reduce latch.");
-
-            latch.get("reduceAwaitLatch").countDown();
-
-            checkStatus(jobId, true);
-
-            assertEquals(10, mapExecCnt.get());
-            assertEquals(0, combineExecCnt.get());
-            assertEquals(1, reduceExecCnt.get());
-        }
-        finally {
-            // Safety.
-            latch.get("mapAwaitLatch").countDown();
-            latch.get("combineAwaitLatch").countDown();
-            latch.get("reduceAwaitLatch").countDown();
-        }
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testTaskWithCombinerPerMap() throws Exception {
-        try {
-            UUID globalId = UUID.randomUUID();
-
-            Job job = Job.getInstance();
-            setupFileSystems(job.getConfiguration());
-
-            job.setMapperClass(TestMapper.class);
-            job.setReducerClass(TestReducer.class);
-            job.setCombinerClass(TestCombiner.class);
-            job.setInputFormatClass(InFormat.class);
-
-            FileOutputFormat.setOutputPath(job, new Path(igfsScheme() + PATH_OUTPUT + "2"));
-
-            GridHadoopJobId jobId = new GridHadoopJobId(globalId, 1);
-
-            grid(0).hadoop().submit(jobId, createJobInfo(job.getConfiguration()));
-
-            checkStatus(jobId, false);
-
-            info("Releasing map latch.");
-
-            latch.get("mapAwaitLatch").countDown();
-
-            checkStatus(jobId, false);
-
-            // All maps are completed. We have a combiner, so no reducers should be executed
-            // before combiner latch is released.
-
-            U.sleep(50);
-
-            assertEquals(0, reduceExecCnt.get());
-
-            info("Releasing combiner latch.");
-
-            latch.get("combineAwaitLatch").countDown();
-
-            checkStatus(jobId, false);
-
-            info("Releasing reduce latch.");
-
-            latch.get("reduceAwaitLatch").countDown();
-
-            checkStatus(jobId, true);
-
-            assertEquals(10, mapExecCnt.get());
-            assertEquals(10, combineExecCnt.get());
-            assertEquals(1, reduceExecCnt.get());
-        }
-        finally {
-            // Safety.
-            latch.get("mapAwaitLatch").countDown();
-            latch.get("combineAwaitLatch").countDown();
-            latch.get("reduceAwaitLatch").countDown();
-        }
-    }
-
-    /**
-     * Checks job execution status.
-     *
-     * @param jobId Job ID.
-     * @param complete Completion status.
-     * @throws Exception If failed.
-     */
-    private void checkStatus(GridHadoopJobId jobId, boolean complete) throws Exception {
-        for (int i = 0; i < gridCount(); i++) {
-            IgniteKernal kernal = (IgniteKernal)grid(i);
-
-            GridHadoop hadoop = kernal.hadoop();
-
-            GridHadoopJobStatus stat = hadoop.status(jobId);
-
-            assert stat != null;
-
-            IgniteInternalFuture<?> fut = hadoop.finishFuture(jobId);
-
-            if (!complete)
-                assertFalse(fut.isDone());
-            else {
-                info("Waiting for status future completion on node [idx=" + i + ", nodeId=" +
-                    kernal.getLocalNodeId() + ']');
-
-                fut.get();
-            }
-        }
-    }
-
-    /**
-     * Test input format
-     */
-    public static class InFormat extends InputFormat {
-
-        @Override public List<InputSplit> getSplits(JobContext ctx) throws IOException, InterruptedException {
-            List<InputSplit> res = new ArrayList<>(BLOCK_CNT);
-
-            for (int i = 0; i < BLOCK_CNT; i++)
-                try {
-                    res.add(new FileSplit(new Path(new URI("someFile")), i, i + 1, new String[] {"localhost"}));
-                }
-                catch (URISyntaxException e) {
-                    throw new IOException(e);
-                }
-
-            return res;
-        }
-
-        @Override public RecordReader createRecordReader(InputSplit split, TaskAttemptContext ctx) throws IOException, InterruptedException {
-            return new RecordReader() {
-                @Override public void initialize(InputSplit split, TaskAttemptContext ctx) {
-                }
-
-                @Override public boolean nextKeyValue() {
-                    return false;
-                }
-
-                @Override public Object getCurrentKey() {
-                    return null;
-                }
-
-                @Override public Object getCurrentValue() {
-                    return null;
-                }
-
-                @Override public float getProgress() {
-                    return 0;
-                }
-
-                @Override public void close() {
-
-                }
-            };
-        }
-    }
-
-    /**
-     * Test mapper.
-     */
-    private static class TestMapper extends Mapper {
-        @Override public void run(Context ctx) throws IOException, InterruptedException {
-            System.out.println("Running task: " + ctx.getTaskAttemptID().getTaskID().getId());
-
-            latch.get("mapAwaitLatch").await();
-
-            mapExecCnt.incrementAndGet();
-
-            System.out.println("Completed task: " + ctx.getTaskAttemptID().getTaskID().getId());
-        }
-    }
-
-    /**
-     * Test reducer.
-     */
-    private static class TestReducer extends Reducer {
-        @Override public void run(Context ctx) throws IOException, InterruptedException {
-            System.out.println("Running task: " + ctx.getTaskAttemptID().getTaskID().getId());
-
-            latch.get("reduceAwaitLatch").await();
-
-            reduceExecCnt.incrementAndGet();
-
-            System.out.println("Completed task: " + ctx.getTaskAttemptID().getTaskID().getId());
-        }
-    }
-
-    /**
-     * Test combiner.
-     */
-    private static class TestCombiner extends Reducer {
-        @Override public void run(Context ctx) throws IOException, InterruptedException {
-            System.out.println("Running task: " + ctx.getTaskAttemptID().getTaskID().getId());
-
-            latch.get("combineAwaitLatch").await();
-
-            combineExecCnt.incrementAndGet();
-
-            System.out.println("Completed task: " + ctx.getTaskAttemptID().getTaskID().getId());
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/6423cf02/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopMapReduceEmbeddedSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopMapReduceEmbeddedSelfTest.java b/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopMapReduceEmbeddedSelfTest.java
deleted file mode 100644
index dda041c..0000000
--- a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopMapReduceEmbeddedSelfTest.java
+++ /dev/null
@@ -1,245 +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 org.apache.hadoop.conf.*;
-import org.apache.hadoop.fs.*;
-import org.apache.hadoop.io.*;
-import org.apache.hadoop.io.serializer.*;
-import org.apache.hadoop.mapred.*;
-import org.apache.hadoop.mapreduce.*;
-import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
-import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
-import org.apache.ignite.igfs.*;
-import org.apache.ignite.internal.*;
-import org.apache.ignite.internal.processors.hadoop.examples.*;
-
-import java.util.*;
-
-import static org.apache.ignite.internal.processors.hadoop.GridHadoopUtils.*;
-
-/**
- * Tests map-reduce execution with embedded mode.
- */
-public class GridHadoopMapReduceEmbeddedSelfTest extends GridHadoopMapReduceTest {
-    /** */
-    private static Map<String, Boolean> flags = GridHadoopSharedMap.map(GridHadoopMapReduceEmbeddedSelfTest.class)
-        .put("flags", new HashMap<String, Boolean>());
-
-    /** {@inheritDoc} */
-    @Override public GridHadoopConfiguration hadoopConfiguration(String gridName) {
-        GridHadoopConfiguration cfg = super.hadoopConfiguration(gridName);
-
-        cfg.setExternalExecution(false);
-
-        return cfg;
-    }
-
-    /**
-     * Tests whole job execution with all phases in old and new versions of API with definition of custom
-     * Serialization, Partitioner and IO formats.
-     * @throws Exception If fails.
-     */
-    public void testMultiReducerWholeMapReduceExecution() throws Exception {
-        IgfsPath inDir = new IgfsPath(PATH_INPUT);
-
-        igfs.mkdirs(inDir);
-
-        IgfsPath inFile = new IgfsPath(inDir, GridHadoopWordCount2.class.getSimpleName() + "-input");
-
-        generateTestFile(inFile.toString(), "key1", 10000, "key2", 20000, "key3", 15000, "key4", 7000, "key5", 12000,
-            "key6", 18000 );
-
-        for (int i = 0; i < 2; i++) {
-            boolean useNewAPI = i == 1;
-
-            igfs.delete(new IgfsPath(PATH_OUTPUT), true);
-
-            flags.put("serializationWasConfigured", false);
-            flags.put("partitionerWasConfigured", false);
-            flags.put("inputFormatWasConfigured", false);
-            flags.put("outputFormatWasConfigured", false);
-
-            JobConf jobConf = new JobConf();
-
-            jobConf.set(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, CustomSerialization.class.getName());
-
-            //To split into about 6-7 items for v2
-            jobConf.setInt(FileInputFormat.SPLIT_MAXSIZE, 65000);
-
-            //For v1
-            jobConf.setInt("fs.local.block.size", 65000);
-
-            // File system coordinates.
-            setupFileSystems(jobConf);
-
-            GridHadoopWordCount1.setTasksClasses(jobConf, !useNewAPI, !useNewAPI, !useNewAPI);
-
-            if (!useNewAPI) {
-                jobConf.setPartitionerClass(CustomV1Partitioner.class);
-                jobConf.setInputFormat(CustomV1InputFormat.class);
-                jobConf.setOutputFormat(CustomV1OutputFormat.class);
-            }
-
-            Job job = Job.getInstance(jobConf);
-
-            GridHadoopWordCount2.setTasksClasses(job, useNewAPI, useNewAPI, useNewAPI);
-
-            if (useNewAPI) {
-                job.setPartitionerClass(CustomV2Partitioner.class);
-                job.setInputFormatClass(CustomV2InputFormat.class);
-                job.setOutputFormatClass(CustomV2OutputFormat.class);
-            }
-
-            job.setOutputKeyClass(Text.class);
-            job.setOutputValueClass(IntWritable.class);
-
-            FileInputFormat.setInputPaths(job, new Path(igfsScheme() + inFile.toString()));
-            FileOutputFormat.setOutputPath(job, new Path(igfsScheme() + PATH_OUTPUT));
-
-            job.setNumReduceTasks(3);
-
-            job.setJarByClass(GridHadoopWordCount2.class);
-
-            IgniteInternalFuture<?> fut = grid(0).hadoop().submit(new GridHadoopJobId(UUID.randomUUID(), 1),
-                    createJobInfo(job.getConfiguration()));
-
-            fut.get();
-
-            assertTrue("Serialization was configured (new API is " + useNewAPI + ")",
-                 flags.get("serializationWasConfigured"));
-
-            assertTrue("Partitioner was configured (new API is = " + useNewAPI + ")",
-                 flags.get("partitionerWasConfigured"));
-
-            assertTrue("Input format was configured (new API is = " + useNewAPI + ")",
-                 flags.get("inputFormatWasConfigured"));
-
-            assertTrue("Output format was configured (new API is = " + useNewAPI + ")",
-                 flags.get("outputFormatWasConfigured"));
-
-            assertEquals("Use new API = " + useNewAPI,
-                "key3\t15000\n" +
-                "key6\t18000\n",
-                readAndSortFile(PATH_OUTPUT + "/" + (useNewAPI ? "part-r-" : "part-") + "00000")
-            );
-
-            assertEquals("Use new API = " + useNewAPI,
-                "key1\t10000\n" +
-                "key4\t7000\n",
-                readAndSortFile(PATH_OUTPUT + "/" + (useNewAPI ? "part-r-" : "part-") + "00001")
-            );
-
-            assertEquals("Use new API = " + useNewAPI,
-                "key2\t20000\n" +
-                "key5\t12000\n",
-                readAndSortFile(PATH_OUTPUT + "/" + (useNewAPI ? "part-r-" : "part-") + "00002")
-            );
-
-        }
-    }
-
-    /**
-     * Custom serialization class that inherits behaviour of native {@link WritableSerialization}.
-     */
-    protected static class CustomSerialization extends WritableSerialization {
-        @Override public void setConf(Configuration conf) {
-            super.setConf(conf);
-
-            flags.put("serializationWasConfigured", true);
-        }
-    }
-
-    /**
-     * Custom implementation of Partitioner in v1 API.
-     */
-    private static class CustomV1Partitioner extends org.apache.hadoop.mapred.lib.HashPartitioner {
-        /** {@inheritDoc} */
-        @Override public void configure(JobConf job) {
-            flags.put("partitionerWasConfigured", true);
-        }
-    }
-
-    /**
-     * Custom implementation of Partitioner in v2 API.
-     */
-    private static class CustomV2Partitioner extends org.apache.hadoop.mapreduce.lib.partition.HashPartitioner
-            implements Configurable {
-        /** {@inheritDoc} */
-        @Override public void setConf(Configuration conf) {
-            flags.put("partitionerWasConfigured", true);
-        }
-
-        /** {@inheritDoc} */
-        @Override public Configuration getConf() {
-            return null;
-        }
-    }
-
-    /**
-     * Custom implementation of InputFormat in v2 API.
-     */
-    private static class CustomV2InputFormat extends org.apache.hadoop.mapreduce.lib.input.TextInputFormat implements Configurable {
-        /** {@inheritDoc} */
-        @Override public void setConf(Configuration conf) {
-            flags.put("inputFormatWasConfigured", true);
-        }
-
-        /** {@inheritDoc} */
-        @Override public Configuration getConf() {
-            return null;
-        }
-    }
-
-    /**
-     * Custom implementation of OutputFormat in v2 API.
-     */
-    private static class CustomV2OutputFormat extends org.apache.hadoop.mapreduce.lib.output.TextOutputFormat implements Configurable {
-        /** {@inheritDoc} */
-        @Override public void setConf(Configuration conf) {
-            flags.put("outputFormatWasConfigured", true);
-        }
-
-        /** {@inheritDoc} */
-        @Override public Configuration getConf() {
-            return null;
-        }
-    }
-
-    /**
-     * Custom implementation of InputFormat in v1 API.
-     */
-    private static class CustomV1InputFormat extends org.apache.hadoop.mapred.TextInputFormat {
-        /** {@inheritDoc} */
-        @Override public void configure(JobConf job) {
-            super.configure(job);
-
-            flags.put("inputFormatWasConfigured", true);
-        }
-    }
-
-    /**
-     * Custom implementation of OutputFormat in v1 API.
-     */
-    private static class CustomV1OutputFormat extends org.apache.hadoop.mapred.TextOutputFormat implements JobConfigurable {
-        /** {@inheritDoc} */
-        @Override public void configure(JobConf job) {
-            flags.put("outputFormatWasConfigured", true);
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/6423cf02/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopMapReduceTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopMapReduceTest.java b/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopMapReduceTest.java
deleted file mode 100644
index 072e764..0000000
--- a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopMapReduceTest.java
+++ /dev/null
@@ -1,195 +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 org.apache.hadoop.fs.*;
-import org.apache.hadoop.io.*;
-import org.apache.hadoop.mapred.*;
-import org.apache.hadoop.mapreduce.*;
-import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
-import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
-import org.apache.ignite.*;
-import org.apache.ignite.igfs.*;
-import org.apache.ignite.internal.*;
-import org.apache.ignite.internal.processors.hadoop.counter.*;
-import org.apache.ignite.internal.processors.hadoop.examples.*;
-import org.apache.ignite.internal.util.lang.*;
-import org.apache.ignite.internal.util.typedef.*;
-import org.apache.ignite.testframework.*;
-
-import java.io.*;
-import java.util.*;
-
-import static org.apache.ignite.internal.processors.hadoop.GridHadoopUtils.*;
-
-/**
- * Test of whole cycle of map-reduce processing via Job tracker.
- */
-public class GridHadoopMapReduceTest extends GridHadoopAbstractWordCountTest {
-    /** {@inheritDoc} */
-    @Override protected int gridCount() {
-        return 3;
-    }
-
-    /**
-     * Tests whole job execution with all phases in all combination of new and old versions of API.
-     * @throws Exception If fails.
-     */
-    public void testWholeMapReduceExecution() throws Exception {
-        IgfsPath inDir = new IgfsPath(PATH_INPUT);
-
-        igfs.mkdirs(inDir);
-
-        IgfsPath inFile = new IgfsPath(inDir, GridHadoopWordCount2.class.getSimpleName() + "-input");
-
-        generateTestFile(inFile.toString(), "red", 100000, "blue", 200000, "green", 150000, "yellow", 70000 );
-
-        for (int i = 0; i < 8; i++) {
-            igfs.delete(new IgfsPath(PATH_OUTPUT), true);
-
-            boolean useNewMapper = (i & 1) == 0;
-            boolean useNewCombiner = (i & 2) == 0;
-            boolean useNewReducer = (i & 4) == 0;
-
-            JobConf jobConf = new JobConf();
-
-            jobConf.set(JOB_COUNTER_WRITER_PROPERTY, GridHadoopFSCounterWriter.class.getName());
-            jobConf.setUser("yyy");
-            jobConf.set(GridHadoopFSCounterWriter.COUNTER_WRITER_DIR_PROPERTY, "/xxx/${USER}/zzz");
-
-            //To split into about 40 items for v2
-            jobConf.setInt(FileInputFormat.SPLIT_MAXSIZE, 65000);
-
-            //For v1
-            jobConf.setInt("fs.local.block.size", 65000);
-
-            // File system coordinates.
-            setupFileSystems(jobConf);
-
-            GridHadoopWordCount1.setTasksClasses(jobConf, !useNewMapper, !useNewCombiner, !useNewReducer);
-
-            Job job = Job.getInstance(jobConf);
-
-            GridHadoopWordCount2.setTasksClasses(job, useNewMapper, useNewCombiner, useNewReducer);
-
-            job.setOutputKeyClass(Text.class);
-            job.setOutputValueClass(IntWritable.class);
-
-            FileInputFormat.setInputPaths(job, new Path(igfsScheme() + inFile.toString()));
-            FileOutputFormat.setOutputPath(job, new Path(igfsScheme() + PATH_OUTPUT));
-
-            job.setJarByClass(GridHadoopWordCount2.class);
-
-            GridHadoopJobId jobId = new GridHadoopJobId(UUID.randomUUID(), 1);
-
-            IgniteInternalFuture<?> fut = grid(0).hadoop().submit(jobId, createJobInfo(job.getConfiguration()));
-
-            fut.get();
-
-            checkJobStatistics(jobId);
-
-            assertEquals("Use new mapper: " + useNewMapper + ", new combiner: " + useNewCombiner + ", new reducer: " +
-                useNewReducer,
-                "blue\t200000\n" +
-                "green\t150000\n" +
-                "red\t100000\n" +
-                "yellow\t70000\n",
-                readAndSortFile(PATH_OUTPUT + "/" + (useNewReducer ? "part-r-" : "part-") + "00000")
-            );
-        }
-    }
-
-    /**
-     * Simple test job statistics.
-     *
-     * @param jobId Job id.
-     * @throws IgniteCheckedException
-     */
-    private void checkJobStatistics(GridHadoopJobId jobId) throws IgniteCheckedException, IOException {
-        GridHadoopCounters cntrs = grid(0).hadoop().counters(jobId);
-
-        GridHadoopPerformanceCounter perfCntr = GridHadoopPerformanceCounter.getCounter(cntrs, null);
-
-        Map<String, SortedMap<Integer,Long>> tasks = new TreeMap<>();
-
-        Map<String, Integer> phaseOrders = new HashMap<>();
-        phaseOrders.put("submit", 0);
-        phaseOrders.put("prepare", 1);
-        phaseOrders.put("start", 2);
-        phaseOrders.put("Cstart", 3);
-        phaseOrders.put("finish", 4);
-
-        String prevTaskId = null;
-
-        long apiEvtCnt = 0;
-
-        for (T2<String, Long> evt : perfCntr.evts()) {
-            //We expect string pattern: COMBINE 1 run 7fa86a14-5a08-40e3-a7cb-98109b52a706
-            String[] parsedEvt = evt.get1().split(" ");
-
-            String taskId;
-            String taskPhase;
-
-            if ("JOB".equals(parsedEvt[0])) {
-                taskId = parsedEvt[0];
-                taskPhase = parsedEvt[1];
-            }
-            else {
-                taskId = ("COMBINE".equals(parsedEvt[0]) ? "MAP" : parsedEvt[0].substring(0, 3)) + parsedEvt[1];
-                taskPhase = ("COMBINE".equals(parsedEvt[0]) ? "C" : "") + parsedEvt[2];
-            }
-
-            if (!taskId.equals(prevTaskId))
-                tasks.put(taskId, new TreeMap<Integer,Long>());
-
-            Integer pos = phaseOrders.get(taskPhase);
-
-            assertNotNull("Invalid phase " + taskPhase, pos);
-
-            tasks.get(taskId).put(pos, evt.get2());
-
-            prevTaskId = taskId;
-
-            apiEvtCnt++;
-        }
-
-        for (Map.Entry<String ,SortedMap<Integer,Long>> task : tasks.entrySet()) {
-            Map<Integer, Long> order = task.getValue();
-
-            long prev = 0;
-
-            for (Map.Entry<Integer, Long> phase : order.entrySet()) {
-                assertTrue("Phase order of " + task.getKey() + " is invalid", phase.getValue() >= prev);
-
-                prev = phase.getValue();
-            }
-        }
-
-        final IgfsPath statPath = new IgfsPath("/xxx/yyy/zzz/" + jobId + "/performance");
-
-        GridTestUtils.waitForCondition(new GridAbsPredicate() {
-            @Override public boolean apply() {
-                return igfs.exists(statPath);
-            }
-        }, 10000);
-
-        BufferedReader reader = new BufferedReader(new InputStreamReader(igfs.open(statPath)));
-
-        assertEquals(apiEvtCnt, GridHadoopTestUtils.simpleCheckJobStatFile(reader));
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/6423cf02/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopPopularWordsTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopPopularWordsTest.java b/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopPopularWordsTest.java
deleted file mode 100644
index 3e8a95a..0000000
--- a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopPopularWordsTest.java
+++ /dev/null
@@ -1,294 +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 com.google.common.collect.*;
-import org.apache.hadoop.conf.*;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.*;
-import org.apache.hadoop.io.*;
-import org.apache.hadoop.mapreduce.*;
-import org.apache.hadoop.mapreduce.lib.input.*;
-import org.apache.hadoop.mapreduce.lib.output.*;
-import org.apache.ignite.internal.util.typedef.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-
-import java.io.*;
-import java.util.*;
-import java.util.Map.*;
-
-import static com.google.common.collect.Maps.*;
-import static com.google.common.collect.MinMaxPriorityQueue.*;
-import static java.util.Collections.*;
-
-/**
- * Hadoop-based 10 popular words example: all files in a given directory are tokenized and for each word longer than
- * 3 characters the number of occurrences ins calculated. Finally, 10 words with the highest occurrence count are
- * output.
- *
- * NOTE: in order to run this example on Windows please ensure that cygwin is installed and available in the system
- * path.
- */
-public class GridHadoopPopularWordsTest {
-    /** Ignite home. */
-    private static final String IGNITE_HOME = U.getIgniteHome();
-
-    /** The path to the input directory. ALl files in that directory will be processed. */
-    private static final Path BOOKS_LOCAL_DIR =
-        new Path("file:" + IGNITE_HOME, "modules/tests/java/org/apache/ignite/grid/hadoop/books");
-
-    /** The path to the output directory. THe result file will be written to this location. */
-    private static final Path RESULT_LOCAL_DIR =
-        new Path("file:" + IGNITE_HOME, "modules/tests/java/org/apache/ignite/grid/hadoop/output");
-
-    /** Popular books source dir in DFS. */
-    private static final Path BOOKS_DFS_DIR = new Path("tmp/word-count-example/in");
-
-    /** Popular books source dir in DFS. */
-    private static final Path RESULT_DFS_DIR = new Path("tmp/word-count-example/out");
-
-    /** Path to the distributed file system configuration. */
-    private static final String DFS_CFG = "examples/config/filesystem/core-site.xml";
-
-    /** Top N words to select **/
-    private static final int POPULAR_WORDS_CNT = 10;
-
-    /**
-     * For each token in the input string the mapper emits a {word, 1} pair.
-     */
-    private static class TokenizingMapper extends Mapper<LongWritable, Text, Text, IntWritable> {
-        /** Constant value. */
-        private static final IntWritable ONE = new IntWritable(1);
-
-        /** The word converted into the Text. */
-        private Text word = new Text();
-
-        /**
-         * Emits a entry where the key is the word and the value is always 1.
-         *
-         * @param key the current position in the input file (not used here)
-         * @param val the text string
-         * @param ctx mapper context
-         * @throws IOException
-         * @throws InterruptedException
-         */
-        @Override protected void map(LongWritable key, Text val, Context ctx)
-            throws IOException, InterruptedException {
-            // Get the mapped object.
-            final String line = val.toString();
-
-            // Splits the given string to words.
-            final String[] words = line.split("[^a-zA-Z0-9]");
-
-            for (final String w : words) {
-                // Only emit counts for longer words.
-                if (w.length() <= 3)
-                    continue;
-
-                word.set(w);
-
-                // Write the word into the context with the initial count equals 1.
-                ctx.write(word, ONE);
-            }
-        }
-    }
-
-    /**
-     * The reducer uses a priority queue to rank the words based on its number of occurrences.
-     */
-    private static class TopNWordsReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
-        private MinMaxPriorityQueue<Entry<Integer, String>> q;
-
-        TopNWordsReducer() {
-            q = orderedBy(reverseOrder(new Comparator<Entry<Integer, String>>() {
-                @Override public int compare(Entry<Integer, String> o1, Entry<Integer, String> o2) {
-                    return o1.getKey().compareTo(o2.getKey());
-                }
-            })).expectedSize(POPULAR_WORDS_CNT).maximumSize(POPULAR_WORDS_CNT).create();
-        }
-
-        /**
-         * This method doesn't emit anything, but just keeps track of the top N words.
-         *
-         * @param key The word.
-         * @param vals The words counts.
-         * @param ctx Reducer context.
-         * @throws IOException If failed.
-         * @throws InterruptedException If failed.
-         */
-        @Override public void reduce(Text key, Iterable<IntWritable> vals, Context ctx) throws IOException,
-            InterruptedException {
-            int sum = 0;
-
-            for (IntWritable val : vals)
-                sum += val.get();
-
-            q.add(immutableEntry(sum, key.toString()));
-        }
-
-        /**
-         * This method is called after all the word entries have been processed. It writes the accumulated
-         * statistics to the job output file.
-         *
-         * @param ctx The job context.
-         * @throws IOException If failed.
-         * @throws InterruptedException If failed.
-         */
-        @Override protected void cleanup(Context ctx) throws IOException, InterruptedException {
-            IntWritable i = new IntWritable();
-
-            Text txt = new Text();
-
-            // iterate in desc order
-            while (!q.isEmpty()) {
-                Entry<Integer, String> e = q.removeFirst();
-
-                i.set(e.getKey());
-
-                txt.set(e.getValue());
-
-                ctx.write(txt, i);
-            }
-        }
-    }
-
-    /**
-     * Configures the Hadoop MapReduce job.
-     *
-     * @return Instance of the Hadoop MapRed job.
-     * @throws IOException If failed.
-     */
-    private Job createConfigBasedHadoopJob() throws IOException {
-        Job jobCfg = new Job();
-
-        Configuration cfg = jobCfg.getConfiguration();
-
-        // Use explicit configuration of distributed file system, if provided.
-        if (DFS_CFG != null)
-            cfg.addResource(U.resolveIgniteUrl(DFS_CFG));
-
-        jobCfg.setJobName("HadoopPopularWordExample");
-        jobCfg.setJarByClass(GridHadoopPopularWordsTest.class);
-        jobCfg.setInputFormatClass(TextInputFormat.class);
-        jobCfg.setOutputKeyClass(Text.class);
-        jobCfg.setOutputValueClass(IntWritable.class);
-        jobCfg.setMapperClass(TokenizingMapper.class);
-        jobCfg.setReducerClass(TopNWordsReducer.class);
-
-        FileInputFormat.setInputPaths(jobCfg, BOOKS_DFS_DIR);
-        FileOutputFormat.setOutputPath(jobCfg, RESULT_DFS_DIR);
-
-        // Local job tracker allows the only task per wave, but text input format
-        // replaces it with the calculated value based on input split size option.
-        if ("local".equals(cfg.get("mapred.job.tracker", "local"))) {
-            // Split job into tasks using 32MB split size.
-            FileInputFormat.setMinInputSplitSize(jobCfg, 32 * 1024 * 1024);
-            FileInputFormat.setMaxInputSplitSize(jobCfg, Long.MAX_VALUE);
-        }
-
-        return jobCfg;
-    }
-
-    /**
-     * Runs the Hadoop job.
-     *
-     * @return {@code True} if succeeded, {@code false} otherwise.
-     * @throws Exception If failed.
-     */
-    private boolean runWordCountConfigBasedHadoopJob() throws Exception {
-        Job job = createConfigBasedHadoopJob();
-
-        // Distributed file system this job will work with.
-        FileSystem fs = FileSystem.get(job.getConfiguration());
-
-        X.println(">>> Using distributed file system: " + fs.getHomeDirectory());
-
-        // Prepare input and output job directories.
-        prepareDirectories(fs);
-
-        long time = System.currentTimeMillis();
-
-        // Run job.
-        boolean res = job.waitForCompletion(true);
-
-        X.println(">>> Job execution time: " + (System.currentTimeMillis() - time) / 1000 + " sec.");
-
-        // Move job results into local file system, so you can view calculated results.
-        publishResults(fs);
-
-        return res;
-    }
-
-    /**
-     * Prepare job's data: cleanup result directories that might have left over
-     * after previous runs, copy input files from the local file system into DFS.
-     *
-     * @param fs Distributed file system to use in job.
-     * @throws IOException If failed.
-     */
-    private void prepareDirectories(FileSystem fs) throws IOException {
-        X.println(">>> Cleaning up DFS result directory: " + RESULT_DFS_DIR);
-
-        fs.delete(RESULT_DFS_DIR, true);
-
-        X.println(">>> Cleaning up DFS input directory: " + BOOKS_DFS_DIR);
-
-        fs.delete(BOOKS_DFS_DIR, true);
-
-        X.println(">>> Copy local files into DFS input directory: " + BOOKS_DFS_DIR);
-
-        fs.copyFromLocalFile(BOOKS_LOCAL_DIR, BOOKS_DFS_DIR);
-    }
-
-    /**
-     * Publish job execution results into local file system, so you can view them.
-     *
-     * @param fs Distributed file sytem used in job.
-     * @throws IOException If failed.
-     */
-    private void publishResults(FileSystem fs) throws IOException {
-        X.println(">>> Cleaning up DFS input directory: " + BOOKS_DFS_DIR);
-
-        fs.delete(BOOKS_DFS_DIR, true);
-
-        X.println(">>> Cleaning up LOCAL result directory: " + RESULT_LOCAL_DIR);
-
-        fs.delete(RESULT_LOCAL_DIR, true);
-
-        X.println(">>> Moving job results into LOCAL result directory: " + RESULT_LOCAL_DIR);
-
-        fs.copyToLocalFile(true, RESULT_DFS_DIR, RESULT_LOCAL_DIR);
-    }
-
-    /**
-     * Executes a modified version of the Hadoop word count example. Here, in addition to counting the number of
-     * occurrences of the word in the source files, the N most popular words are selected.
-     *
-     * @param args None.
-     */
-    public static void main(String[] args) {
-        try {
-            new GridHadoopPopularWordsTest().runWordCountConfigBasedHadoopJob();
-        }
-        catch (Exception e) {
-            X.println(">>> Failed to run word count example: " + e.getMessage());
-        }
-
-        System.exit(0);
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/6423cf02/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopSerializationWrapperSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopSerializationWrapperSelfTest.java b/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopSerializationWrapperSelfTest.java
deleted file mode 100644
index 79b9965..0000000
--- a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopSerializationWrapperSelfTest.java
+++ /dev/null
@@ -1,74 +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 org.apache.hadoop.io.*;
-import org.apache.hadoop.io.serializer.*;
-import org.apache.ignite.internal.processors.hadoop.v2.*;
-import org.apache.ignite.testframework.junits.common.*;
-
-import java.io.*;
-import java.util.*;
-
-/**
- * Test of wrapper of the native serialization.
- */
-public class GridHadoopSerializationWrapperSelfTest extends GridCommonAbstractTest {
-    /**
-     * Tests read/write of IntWritable via native WritableSerialization.
-     * @throws Exception If fails.
-     */
-    public void testIntWritableSerialization() throws Exception {
-        GridHadoopSerialization ser = new GridHadoopSerializationWrapper(new WritableSerialization(), IntWritable.class);
-
-        ByteArrayOutputStream buf = new ByteArrayOutputStream();
-
-        DataOutput out = new DataOutputStream(buf);
-
-        ser.write(out, new IntWritable(3));
-        ser.write(out, new IntWritable(-5));
-
-        assertEquals("[0, 0, 0, 3, -1, -1, -1, -5]", Arrays.toString(buf.toByteArray()));
-
-        DataInput in = new DataInputStream(new ByteArrayInputStream(buf.toByteArray()));
-
-        assertEquals(3, ((IntWritable)ser.read(in, null)).get());
-        assertEquals(-5, ((IntWritable)ser.read(in, null)).get());
-    }
-
-    /**
-     * Tests read/write of Integer via native JavaleSerialization.
-     * @throws Exception If fails.
-     */
-    public void testIntJavaSerialization() throws Exception {
-        GridHadoopSerialization ser = new GridHadoopSerializationWrapper(new JavaSerialization(), Integer.class);
-
-        ByteArrayOutputStream buf = new ByteArrayOutputStream();
-
-        DataOutput out = new DataOutputStream(buf);
-
-        ser.write(out, 3);
-        ser.write(out, -5);
-        ser.close();
-
-        DataInput in = new DataInputStream(new ByteArrayInputStream(buf.toByteArray()));
-
-        assertEquals(3, ((Integer)ser.read(in, null)).intValue());
-        assertEquals(-5, ((Integer)ser.read(in, null)).intValue());
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/6423cf02/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopSharedMap.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopSharedMap.java b/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopSharedMap.java
deleted file mode 100644
index 689fb58..0000000
--- a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopSharedMap.java
+++ /dev/null
@@ -1,67 +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 org.jdk8.backport.*;
-
-import java.util.concurrent.*;
-
-/**
- * For tests.
- */
-public class GridHadoopSharedMap {
-    /** */
-    private static final ConcurrentMap<String, GridHadoopSharedMap> maps = new ConcurrentHashMap8<>();
-
-    /** */
-    private final ConcurrentMap<String, Object> map = new ConcurrentHashMap8<>();
-
-    /**
-     * Private.
-     */
-    private GridHadoopSharedMap() {
-        // No-op.
-    }
-
-    /**
-     * Puts object by key.
-     *
-     * @param key Key.
-     * @param val Value.
-     */
-    public <T> T put(String key, T val) {
-        Object old = map.putIfAbsent(key, val);
-
-        return old == null ? val : (T)old;
-    }
-
-    /**
-     * @param cls Class.
-     * @return Map of static fields.
-     */
-    public static GridHadoopSharedMap map(Class<?> cls) {
-        GridHadoopSharedMap m = maps.get(cls.getName());
-
-        if (m != null)
-            return m;
-
-        GridHadoopSharedMap old = maps.putIfAbsent(cls.getName(), m = new GridHadoopSharedMap());
-
-        return old == null ? m : old;
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/6423cf02/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopSortingExternalTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopSortingExternalTest.java b/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopSortingExternalTest.java
deleted file mode 100644
index 23884ef..0000000
--- a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopSortingExternalTest.java
+++ /dev/null
@@ -1,32 +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;
-
-/**
- * External test for sorting.
- */
-public class GridHadoopSortingExternalTest extends GridHadoopSortingTest {
-    /** {@inheritDoc} */
-    @Override public GridHadoopConfiguration hadoopConfiguration(String gridName) {
-        GridHadoopConfiguration cfg = super.hadoopConfiguration(gridName);
-
-        cfg.setExternalExecution(true);
-
-        return cfg;
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/6423cf02/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopSortingTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopSortingTest.java b/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopSortingTest.java
deleted file mode 100644
index 3a2c397..0000000
--- a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopSortingTest.java
+++ /dev/null
@@ -1,281 +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 org.apache.hadoop.fs.*;
-import org.apache.hadoop.io.*;
-import org.apache.hadoop.io.serializer.*;
-import org.apache.hadoop.mapreduce.*;
-import org.apache.hadoop.mapreduce.lib.input.*;
-import org.apache.hadoop.mapreduce.lib.output.*;
-import org.apache.ignite.internal.util.typedef.*;
-
-import java.io.*;
-import java.net.*;
-import java.util.*;
-
-import static org.apache.ignite.internal.processors.hadoop.GridHadoopUtils.*;
-
-/**
- * Tests correct sorting.
- */
-public class GridHadoopSortingTest extends GridHadoopAbstractSelfTest {
-    /** */
-    private static final String PATH_INPUT = "/test-in";
-
-    /** */
-    private static final String PATH_OUTPUT = "/test-out";
-
-    /** {@inheritDoc} */
-    @Override protected int gridCount() {
-        return 3;
-    }
-
-    /**
-     * @return {@code True} if IGFS is enabled on Hadoop nodes.
-     */
-    @Override protected boolean igfsEnabled() {
-        return true;
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void beforeTest() throws Exception {
-        startGrids(gridCount());
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void afterTest() throws Exception {
-        stopAllGrids(true);
-    }
-
-    /** {@inheritDoc} */
-    @Override public GridHadoopConfiguration hadoopConfiguration(String gridName) {
-        GridHadoopConfiguration cfg = super.hadoopConfiguration(gridName);
-
-        cfg.setExternalExecution(false);
-
-        return cfg;
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testSortSimple() throws Exception {
-        // Generate test data.
-        Job job = Job.getInstance();
-
-        job.setInputFormatClass(InFormat.class);
-
-        job.setOutputKeyClass(Text.class);
-        job.setOutputValueClass(NullWritable.class);
-
-        job.setMapperClass(Mapper.class);
-        job.setNumReduceTasks(0);
-
-        setupFileSystems(job.getConfiguration());
-
-        FileOutputFormat.setOutputPath(job, new Path(igfsScheme() + PATH_INPUT));
-
-        X.printerrln("Data generation started.");
-
-        grid(0).hadoop().submit(new GridHadoopJobId(UUID.randomUUID(), 1),
-            createJobInfo(job.getConfiguration())).get(180000);
-
-        X.printerrln("Data generation complete.");
-
-        // Run main map-reduce job.
-        job = Job.getInstance();
-
-        setupFileSystems(job.getConfiguration());
-
-        job.getConfiguration().set(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY, JavaSerialization.class.getName() +
-            "," + WritableSerialization.class.getName());
-
-        FileInputFormat.setInputPaths(job, new Path(igfsScheme() + PATH_INPUT));
-        FileOutputFormat.setOutputPath(job, new Path(igfsScheme() + PATH_OUTPUT));
-
-        job.setSortComparatorClass(JavaSerializationComparator.class);
-
-        job.setMapperClass(MyMapper.class);
-        job.setReducerClass(MyReducer.class);
-
-        job.setNumReduceTasks(2);
-
-        job.setMapOutputKeyClass(UUID.class);
-        job.setMapOutputValueClass(NullWritable.class);
-
-        job.setOutputKeyClass(Text.class);
-        job.setOutputValueClass(NullWritable.class);
-
-        X.printerrln("Job started.");
-
-        grid(0).hadoop().submit(new GridHadoopJobId(UUID.randomUUID(), 2),
-            createJobInfo(job.getConfiguration())).get(180000);
-
-        X.printerrln("Job complete.");
-
-        // Check result.
-        Path outDir = new Path(igfsScheme() + PATH_OUTPUT);
-
-        AbstractFileSystem fs = AbstractFileSystem.get(new URI(igfsScheme()), job.getConfiguration());
-
-        for (FileStatus file : fs.listStatus(outDir)) {
-            X.printerrln("__ file: " + file);
-
-            if (file.getLen() == 0)
-                continue;
-
-            FSDataInputStream in = fs.open(file.getPath());
-
-            Scanner sc = new Scanner(in);
-
-            UUID prev = null;
-
-            while(sc.hasNextLine()) {
-                UUID next = UUID.fromString(sc.nextLine());
-
-//                X.printerrln("___ check: " + next);
-
-                if (prev != null)
-                    assertTrue(prev.compareTo(next) < 0);
-
-                prev = next;
-            }
-        }
-    }
-
-    public static class InFormat extends InputFormat<Text, NullWritable> {
-        /** {@inheritDoc} */
-        @Override public List<InputSplit> getSplits(JobContext ctx) throws IOException, InterruptedException {
-            List<InputSplit> res = new ArrayList<>();
-
-            FakeSplit split = new FakeSplit(20);
-
-            for (int i = 0; i < 10; i++)
-                res.add(split);
-
-            return res;
-        }
-
-        /** {@inheritDoc} */
-        @Override public RecordReader<Text, NullWritable> createRecordReader(final InputSplit split,
-            TaskAttemptContext ctx) throws IOException, InterruptedException {
-            return new RecordReader<Text, NullWritable>() {
-                /** */
-                int cnt;
-
-                /** */
-                Text txt = new Text();
-
-                @Override public void initialize(InputSplit split, TaskAttemptContext ctx) {
-                    // No-op.
-                }
-
-                @Override public boolean nextKeyValue() throws IOException, InterruptedException {
-                    return ++cnt <= split.getLength();
-                }
-
-                @Override public Text getCurrentKey() {
-                    txt.set(UUID.randomUUID().toString());
-
-//                    X.printerrln("___ read: " + txt);
-
-                    return txt;
-                }
-
-                @Override public NullWritable getCurrentValue() {
-                    return NullWritable.get();
-                }
-
-                @Override public float getProgress() throws IOException, InterruptedException {
-                    return (float)cnt / split.getLength();
-                }
-
-                @Override public void close() {
-                    // No-op.
-                }
-            };
-        }
-    }
-
-    public static class MyMapper extends Mapper<LongWritable, Text, UUID, NullWritable> {
-        /** {@inheritDoc} */
-        @Override protected void map(LongWritable key, Text val, Context ctx) throws IOException, InterruptedException {
-//            X.printerrln("___ map: " + val);
-
-            ctx.write(UUID.fromString(val.toString()), NullWritable.get());
-        }
-    }
-
-    public static class MyReducer extends Reducer<UUID, NullWritable, Text, NullWritable> {
-        /** */
-        private Text text = new Text();
-
-        /** {@inheritDoc} */
-        @Override protected void reduce(UUID key, Iterable<NullWritable> vals, Context ctx)
-            throws IOException, InterruptedException {
-//            X.printerrln("___ rdc: " + key);
-
-            text.set(key.toString());
-
-            ctx.write(text, NullWritable.get());
-        }
-    }
-
-    public static class FakeSplit extends InputSplit implements Writable {
-        /** */
-        private static final String[] HOSTS = {"127.0.0.1"};
-
-        /** */
-        private int len;
-
-        /**
-         * @param len Length.
-         */
-        public FakeSplit(int len) {
-            this.len = len;
-        }
-
-        /**
-         *
-         */
-        public FakeSplit() {
-            // No-op.
-        }
-
-        /** {@inheritDoc} */
-        @Override public long getLength() throws IOException, InterruptedException {
-            return len;
-        }
-
-        /** {@inheritDoc} */
-        @Override public String[] getLocations() throws IOException, InterruptedException {
-            return HOSTS;
-        }
-
-        /** {@inheritDoc} */
-        @Override public void write(DataOutput out) throws IOException {
-            out.writeInt(len);
-        }
-
-        /** {@inheritDoc} */
-        @Override public void readFields(DataInput in) throws IOException {
-            len = in.readInt();
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/6423cf02/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopSplitWrapperSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopSplitWrapperSelfTest.java b/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopSplitWrapperSelfTest.java
deleted file mode 100644
index 0b15a2c..0000000
--- a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopSplitWrapperSelfTest.java
+++ /dev/null
@@ -1,68 +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 org.apache.hadoop.fs.*;
-import org.apache.hadoop.mapreduce.lib.input.*;
-import org.apache.ignite.internal.processors.hadoop.v2.*;
-import org.apache.ignite.testframework.*;
-
-import java.io.*;
-import java.util.*;
-import java.util.concurrent.*;
-
-/**
- * Self test of {@link GridHadoopSplitWrapper}.
- */
-public class GridHadoopSplitWrapperSelfTest extends GridHadoopAbstractSelfTest {
-    /**
-     * Tests serialization of wrapper and the wrapped native split.
-     * @throws Exception If fails.
-     */
-    public void testSerialization() throws Exception {
-        FileSplit nativeSplit = new FileSplit(new Path("/path/to/file"), 100, 500, new String[]{"host1", "host2"});
-
-        assertEquals("/path/to/file:100+500", nativeSplit.toString());
-
-        GridHadoopSplitWrapper split = GridHadoopUtils.wrapSplit(10, nativeSplit, nativeSplit.getLocations());
-
-        assertEquals("[host1, host2]", Arrays.toString(split.hosts()));
-
-        ByteArrayOutputStream buf = new ByteArrayOutputStream();
-
-        ObjectOutput out = new ObjectOutputStream(buf);
-
-        out.writeObject(split);
-
-        ObjectInput in = new ObjectInputStream(new ByteArrayInputStream(buf.toByteArray()));
-
-        final GridHadoopSplitWrapper res = (GridHadoopSplitWrapper)in.readObject();
-
-        assertEquals("/path/to/file:100+500", GridHadoopUtils.unwrapSplit(res).toString());
-
-        GridTestUtils.assertThrows(log, new Callable<Object>() {
-            @Override public Object call() throws Exception {
-                res.hosts();
-
-                return null;
-            }
-        }, AssertionError.class, null);
-    }
-
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/6423cf02/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopStartup.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopStartup.java b/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopStartup.java
deleted file mode 100644
index 6cc7635..0000000
--- a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopStartup.java
+++ /dev/null
@@ -1,55 +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 org.apache.hadoop.conf.*;
-import org.apache.ignite.*;
-import org.apache.ignite.igfs.hadoop.v2.IgfsHadoopFileSystem;
-import org.apache.ignite.internal.util.typedef.*;
-
-/**
- * Hadoop node startup.
- */
-public class GridHadoopStartup {
-    /**
-     * @param args Arguments.
-     */
-    public static void main(String[] args) {
-        G.start("config/hadoop/default-config.xml");
-    }
-
-    /**
-     * @return Configuration for job run.
-     */
-    @SuppressWarnings("UnnecessaryFullyQualifiedName")
-    public static Configuration configuration() {
-        Configuration cfg = new Configuration();
-
-        cfg.set("fs.defaultFS", "igfs://igfs@localhost");
-
-        cfg.set("fs.igfs.impl", org.apache.ignite.igfs.hadoop.v1.IgfsHadoopFileSystem.class.getName());
-        cfg.set("fs.AbstractFileSystem.igfs.impl", IgfsHadoopFileSystem.class.getName());
-
-        cfg.set("dfs.client.block.write.replace-datanode-on-failure.policy", "NEVER");
-
-        cfg.set("mapreduce.framework.name", "ignite");
-        cfg.set("mapreduce.jobtracker.address", "localhost:11211");
-
-        return cfg;
-    }
-}

http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/6423cf02/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopTaskExecutionSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopTaskExecutionSelfTest.java b/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopTaskExecutionSelfTest.java
deleted file mode 100644
index 40546bb..0000000
--- a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/GridHadoopTaskExecutionSelfTest.java
+++ /dev/null
@@ -1,551 +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 org.apache.hadoop.conf.*;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.*;
-import org.apache.hadoop.io.*;
-import org.apache.hadoop.mapreduce.*;
-import org.apache.hadoop.mapreduce.lib.input.*;
-import org.apache.hadoop.mapreduce.lib.output.*;
-import org.apache.ignite.*;
-import org.apache.ignite.configuration.*;
-import org.apache.ignite.igfs.*;
-import org.apache.ignite.igfs.hadoop.v1.*;
-import org.apache.ignite.internal.*;
-import org.apache.ignite.internal.util.lang.*;
-import org.apache.ignite.internal.util.typedef.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-import org.apache.ignite.testframework.*;
-
-import java.io.*;
-import java.util.*;
-import java.util.concurrent.*;
-import java.util.concurrent.atomic.*;
-
-import static org.apache.ignite.internal.processors.hadoop.GridHadoopUtils.*;
-
-/**
- * Tests map-reduce task execution basics.
- */
-public class GridHadoopTaskExecutionSelfTest extends GridHadoopAbstractSelfTest {
-    /** */
-    private static GridHadoopSharedMap m = GridHadoopSharedMap.map(GridHadoopTaskExecutionSelfTest.class);
-
-    /** Line count. */
-    private static final AtomicInteger totalLineCnt = m.put("totalLineCnt", new AtomicInteger());
-
-    /** Executed tasks. */
-    private static final AtomicInteger executedTasks = m.put("executedTasks", new AtomicInteger());
-
-    /** Cancelled tasks. */
-    private static final AtomicInteger cancelledTasks = m.put("cancelledTasks", new AtomicInteger());
-
-    /** Working directory of each task. */
-    private static final Map<String, String> taskWorkDirs = m.put("taskWorkDirs",
-        new ConcurrentHashMap<String, String>());
-
-    /** Mapper id to fail. */
-    private static final AtomicInteger failMapperId = m.put("failMapperId", new AtomicInteger());
-
-    /** Number of splits of the current input. */
-    private static final AtomicInteger splitsCount = m.put("splitsCount", new AtomicInteger());
-
-    /** Test param. */
-    private static final String MAP_WRITE = "test.map.write";
-
-
-    /** {@inheritDoc} */
-    @Override public IgfsConfiguration igfsConfiguration() {
-        IgfsConfiguration cfg = super.igfsConfiguration();
-
-        cfg.setFragmentizerEnabled(false);
-
-        return cfg;
-    }
-
-    /** {@inheritDoc} */
-    @Override protected boolean igfsEnabled() {
-        return true;
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void beforeTestsStarted() throws Exception {
-        super.beforeTestsStarted();
-
-        startGrids(gridCount());
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void afterTestsStopped() throws Exception {
-        stopAllGrids();
-
-        super.afterTestsStopped();
-    }
-
-    /** {@inheritDoc} */
-    @Override protected void beforeTest() throws Exception {
-        grid(0).fileSystem(igfsName).format();
-    }
-
-    /** {@inheritDoc} */
-    @Override public GridHadoopConfiguration hadoopConfiguration(String gridName) {
-        GridHadoopConfiguration cfg = super.hadoopConfiguration(gridName);
-
-        cfg.setMaxParallelTasks(5);
-        cfg.setExternalExecution(false);
-
-        return cfg;
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testMapRun() throws Exception {
-        int lineCnt = 10000;
-        String fileName = "/testFile";
-
-        prepareFile(fileName, lineCnt);
-
-        totalLineCnt.set(0);
-        taskWorkDirs.clear();
-
-        Configuration cfg = new Configuration();
-
-        cfg.setStrings("fs.igfs.impl", IgfsHadoopFileSystem.class.getName());
-
-        Job job = Job.getInstance(cfg);
-        job.setOutputKeyClass(Text.class);
-        job.setOutputValueClass(IntWritable.class);
-
-        job.setMapperClass(TestMapper.class);
-
-        job.setNumReduceTasks(0);
-
-        job.setInputFormatClass(TextInputFormat.class);
-
-        FileInputFormat.setInputPaths(job, new Path("igfs://:" + getTestGridName(0) + "@/"));
-        FileOutputFormat.setOutputPath(job, new Path("igfs://:" + getTestGridName(0) + "@/output/"));
-
-        job.setJarByClass(getClass());
-
-        IgniteInternalFuture<?> fut = grid(0).hadoop().submit(new GridHadoopJobId(UUID.randomUUID(), 1),
-                createJobInfo(job.getConfiguration()));
-
-        fut.get();
-
-        assertEquals(lineCnt, totalLineCnt.get());
-
-        assertEquals(32, taskWorkDirs.size());
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testMapCombineRun() throws Exception {
-        int lineCnt = 10001;
-        String fileName = "/testFile";
-
-        prepareFile(fileName, lineCnt);
-
-        totalLineCnt.set(0);
-        taskWorkDirs.clear();
-
-        Configuration cfg = new Configuration();
-
-        cfg.setStrings("fs.igfs.impl", IgfsHadoopFileSystem.class.getName());
-        cfg.setBoolean(MAP_WRITE, true);
-
-        Job job = Job.getInstance(cfg);
-        job.setOutputKeyClass(Text.class);
-        job.setOutputValueClass(IntWritable.class);
-
-        job.setMapperClass(TestMapper.class);
-        job.setCombinerClass(TestCombiner.class);
-        job.setReducerClass(TestReducer.class);
-
-        job.setNumReduceTasks(2);
-
-        job.setInputFormatClass(TextInputFormat.class);
-
-        FileInputFormat.setInputPaths(job, new Path("igfs://:" + getTestGridName(0) + "@/"));
-        FileOutputFormat.setOutputPath(job, new Path("igfs://:" + getTestGridName(0) + "@/output"));
-
-        job.setJarByClass(getClass());
-
-        GridHadoopJobId jobId = new GridHadoopJobId(UUID.randomUUID(), 2);
-
-        IgniteInternalFuture<?> fut = grid(0).hadoop().submit(jobId, createJobInfo(job.getConfiguration()));
-
-        fut.get();
-
-        assertEquals(lineCnt, totalLineCnt.get());
-
-        assertEquals(34, taskWorkDirs.size());
-
-        for (int g = 0; g < gridCount(); g++)
-            grid(g).hadoop().finishFuture(jobId).get();
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testMapperException() throws Exception {
-        prepareFile("/testFile", 1000);
-
-        Configuration cfg = new Configuration();
-
-        cfg.setStrings("fs.igfs.impl", IgfsHadoopFileSystem.class.getName());
-
-        Job job = Job.getInstance(cfg);
-        job.setOutputKeyClass(Text.class);
-        job.setOutputValueClass(IntWritable.class);
-
-        job.setMapperClass(FailMapper.class);
-
-        job.setNumReduceTasks(0);
-
-        job.setInputFormatClass(TextInputFormat.class);
-
-        FileInputFormat.setInputPaths(job, new Path("igfs://:" + getTestGridName(0) + "@/"));
-        FileOutputFormat.setOutputPath(job, new Path("igfs://:" + getTestGridName(0) + "@/output/"));
-
-        job.setJarByClass(getClass());
-
-        final IgniteInternalFuture<?> fut = grid(0).hadoop().submit(new GridHadoopJobId(UUID.randomUUID(), 3),
-                createJobInfo(job.getConfiguration()));
-
-        GridTestUtils.assertThrows(log, new Callable<Object>() {
-            @Override public Object call() throws Exception {
-                fut.get();
-
-                return null;
-            }
-        }, IgniteCheckedException.class, null);
-    }
-
-    /**
-     * @param fileName File name.
-     * @param lineCnt Line count.
-     * @throws Exception If failed.
-     */
-    private void prepareFile(String fileName, int lineCnt) throws Exception {
-        IgniteFs igfs = grid(0).fileSystem(igfsName);
-
-        try (OutputStream os = igfs.create(new IgfsPath(fileName), true)) {
-            PrintWriter w = new PrintWriter(new OutputStreamWriter(os));
-
-            for (int i = 0; i < lineCnt; i++)
-                w.print("Hello, Hadoop map-reduce!\n");
-
-            w.flush();
-        }
-    }
-
-    /**
-     * Prepare job with mappers to cancel.
-     * @return Fully configured job.
-     * @throws Exception If fails.
-     */
-    private Configuration prepareJobForCancelling() throws Exception {
-        prepareFile("/testFile", 1500);
-
-        executedTasks.set(0);
-        cancelledTasks.set(0);
-        failMapperId.set(0);
-        splitsCount.set(0);
-
-        Configuration cfg = new Configuration();
-
-        setupFileSystems(cfg);
-
-        Job job = Job.getInstance(cfg);
-        job.setOutputKeyClass(Text.class);
-        job.setOutputValueClass(IntWritable.class);
-
-        job.setMapperClass(CancellingTestMapper.class);
-
-        job.setNumReduceTasks(0);
-
-        job.setInputFormatClass(InFormat.class);
-
-        FileInputFormat.setInputPaths(job, new Path("igfs://:" + getTestGridName(0) + "@/"));
-        FileOutputFormat.setOutputPath(job, new Path("igfs://:" + getTestGridName(0) + "@/output/"));
-
-        job.setJarByClass(getClass());
-
-        return job.getConfiguration();
-    }
-
-    /**
-     * Test input format.
-     */
-    private static class InFormat extends TextInputFormat {
-        @Override public List<InputSplit> getSplits(JobContext ctx) throws IOException {
-            List<InputSplit> res = super.getSplits(ctx);
-
-            splitsCount.set(res.size());
-
-            X.println("___ split of input: " + splitsCount.get());
-
-            return res;
-        }
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testTaskCancelling() throws Exception {
-        Configuration cfg = prepareJobForCancelling();
-
-        GridHadoopJobId jobId = new GridHadoopJobId(UUID.randomUUID(), 1);
-
-        final IgniteInternalFuture<?> fut = grid(0).hadoop().submit(jobId, createJobInfo(cfg));
-
-        if (!GridTestUtils.waitForCondition(new GridAbsPredicate() {
-            @Override public boolean apply() {
-                return splitsCount.get() > 0;
-            }
-        }, 20000)) {
-            U.dumpThreads(log);
-
-            assertTrue(false);
-        }
-
-        if (!GridTestUtils.waitForCondition(new GridAbsPredicate() {
-            @Override public boolean apply() {
-                return executedTasks.get() == splitsCount.get();
-            }
-        }, 20000)) {
-            U.dumpThreads(log);
-
-            assertTrue(false);
-        }
-
-        // Fail mapper with id "1", cancels others
-        failMapperId.set(1);
-
-        GridTestUtils.assertThrows(log, new Callable<Object>() {
-            @Override public Object call() throws Exception {
-                fut.get();
-
-                return null;
-            }
-        }, IgniteCheckedException.class, null);
-
-        assertEquals(executedTasks.get(), cancelledTasks.get() + 1);
-    }
-
-    /**
-     * @throws Exception If failed.
-     */
-    public void testJobKill() throws Exception {
-        Configuration cfg = prepareJobForCancelling();
-
-        GridHadoop hadoop = grid(0).hadoop();
-
-        GridHadoopJobId jobId = new GridHadoopJobId(UUID.randomUUID(), 1);
-
-        //Kill unknown job.
-        boolean killRes = hadoop.kill(jobId);
-
-        assertFalse(killRes);
-
-        final IgniteInternalFuture<?> fut = hadoop.submit(jobId, createJobInfo(cfg));
-
-        if (!GridTestUtils.waitForCondition(new GridAbsPredicate() {
-            @Override public boolean apply() {
-                return splitsCount.get() > 0;
-            }
-        }, 20000)) {
-            U.dumpThreads(log);
-
-            assertTrue(false);
-        }
-
-        if (!GridTestUtils.waitForCondition(new GridAbsPredicate() {
-            @Override public boolean apply() {
-                X.println("___ executed tasks: " + executedTasks.get());
-
-                return executedTasks.get() == splitsCount.get();
-            }
-        }, 20000)) {
-            U.dumpThreads(log);
-
-            fail();
-        }
-
-        //Kill really ran job.
-        killRes = hadoop.kill(jobId);
-
-        assertTrue(killRes);
-
-        GridTestUtils.assertThrows(log, new Callable<Object>() {
-            @Override public Object call() throws Exception {
-                fut.get();
-
-                return null;
-            }
-        }, IgniteCheckedException.class, null);
-
-        assertEquals(executedTasks.get(), cancelledTasks.get());
-
-        //Kill the same job again.
-        killRes = hadoop.kill(jobId);
-
-        assertTrue(killRes);
-    }
-
-    private static class CancellingTestMapper extends Mapper<Object, Text, Text, IntWritable> {
-        private int mapperId;
-
-        /** {@inheritDoc} */
-        @Override protected void setup(Context ctx) throws IOException, InterruptedException {
-            mapperId = executedTasks.incrementAndGet();
-        }
-
-        /** {@inheritDoc} */
-        @Override public void run(Context ctx) throws IOException, InterruptedException {
-            try {
-                super.run(ctx);
-            }
-            catch (GridHadoopTaskCancelledException e) {
-                cancelledTasks.incrementAndGet();
-
-                throw e;
-            }
-        }
-
-        /** {@inheritDoc} */
-        @Override protected void map(Object key, Text val, Context ctx) throws IOException, InterruptedException {
-            if (mapperId == failMapperId.get())
-                throw new IOException();
-
-            Thread.sleep(1000);
-        }
-    }
-
-    /**
-     * Test failing mapper.
-     */
-    private static class FailMapper extends Mapper<Object, Text, Text, IntWritable> {
-        /** {@inheritDoc} */
-        @Override protected void map(Object key, Text val, Context ctx) throws IOException, InterruptedException {
-            throw new IOException("Expected");
-        }
-    }
-
-    /**
-     * Mapper calculates number of lines.
-     */
-    private static class TestMapper extends Mapper<Object, Text, Text, IntWritable> {
-        /** Writable integer constant of '1'. */
-        private static final IntWritable ONE = new IntWritable(1);
-
-        /** Line count constant. */
-        public static final Text LINE_COUNT = new Text("lineCount");
-
-        /** {@inheritDoc} */
-        @Override protected void setup(Context ctx) throws IOException, InterruptedException {
-            X.println("___ Mapper: " + ctx.getTaskAttemptID());
-
-            String taskId = ctx.getTaskAttemptID().toString();
-
-            LocalFileSystem locFs = FileSystem.getLocal(ctx.getConfiguration());
-
-            String workDir = locFs.getWorkingDirectory().toString();
-
-            assertNull(taskWorkDirs.put(workDir, taskId));
-        }
-
-        /** {@inheritDoc} */
-        @Override protected void map(Object key, Text val, Context ctx) throws IOException, InterruptedException {
-            if (ctx.getConfiguration().getBoolean(MAP_WRITE, false))
-                ctx.write(LINE_COUNT, ONE);
-            else
-                totalLineCnt.incrementAndGet();
-        }
-    }
-
-    /**
-     * Combiner calculates number of lines.
-     */
-    private static class TestCombiner extends Reducer<Text, IntWritable, Text, IntWritable> {
-        /** */
-        IntWritable sum = new IntWritable();
-
-        /** {@inheritDoc} */
-        @Override protected void setup(Context ctx) throws IOException, InterruptedException {
-            X.println("___ Combiner: ");
-        }
-
-        /** {@inheritDoc} */
-        @Override protected void reduce(Text key, Iterable<IntWritable> values, Context ctx) throws IOException,
-            InterruptedException {
-            int lineCnt = 0;
-
-            for (IntWritable value : values)
-                lineCnt += value.get();
-
-            sum.set(lineCnt);
-
-            X.println("___ combo: " + lineCnt);
-
-            ctx.write(key, sum);
-        }
-    }
-
-    /**
-     * Combiner calculates number of lines.
-     */
-    private static class TestReducer extends Reducer<Text, IntWritable, Text, IntWritable> {
-        /** */
-        IntWritable sum = new IntWritable();
-
-        /** {@inheritDoc} */
-        @Override protected void setup(Context ctx) throws IOException, InterruptedException {
-            X.println("___ Reducer: " + ctx.getTaskAttemptID());
-
-            String taskId = ctx.getTaskAttemptID().toString();
-            String workDir = FileSystem.getLocal(ctx.getConfiguration()).getWorkingDirectory().toString();
-
-            assertNull(taskWorkDirs.put(workDir, taskId));
-        }
-
-        /** {@inheritDoc} */
-        @Override protected void reduce(Text key, Iterable<IntWritable> values, Context ctx) throws IOException,
-            InterruptedException {
-            int lineCnt = 0;
-
-            for (IntWritable value : values) {
-                lineCnt += value.get();
-
-                X.println("___ rdcr: " + value.get());
-            }
-
-            sum.set(lineCnt);
-
-            ctx.write(key, sum);
-
-            X.println("___ RDCR SUM: " + lineCnt);
-
-            totalLineCnt.addAndGet(lineCnt);
-        }
-    }
-}


Mime
View raw message