You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2016/09/27 15:26:15 UTC
[25/68] [abbrv] ignite git commit: IGNITE-3912: Hadoop: Implemented
new class loading architecture for embedded execution mode.
http://git-wip-us.apache.org/repos/asf/ignite/blob/8032fc2c/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopDefaultMapReducePlannerSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopDefaultMapReducePlannerSelfTest.java b/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopDefaultMapReducePlannerSelfTest.java
deleted file mode 100644
index a69b72a..0000000
--- a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopDefaultMapReducePlannerSelfTest.java
+++ /dev/null
@@ -1,615 +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.ignite.IgniteCheckedException;
-import org.apache.ignite.IgniteFileSystem;
-import org.apache.ignite.cluster.ClusterNode;
-import org.apache.ignite.hadoop.mapreduce.IgniteHadoopMapReducePlanner;
-import org.apache.ignite.igfs.IgfsBlockLocation;
-import org.apache.ignite.igfs.IgfsPath;
-import org.apache.ignite.internal.processors.hadoop.planner.HadoopAbstractMapReducePlanner;
-import org.apache.ignite.internal.processors.igfs.IgfsBlockLocationImpl;
-import org.apache.ignite.internal.processors.igfs.IgfsIgniteMock;
-import org.apache.ignite.internal.processors.igfs.IgfsMock;
-import org.apache.ignite.internal.util.typedef.F;
-import org.apache.ignite.testframework.GridTestNode;
-import org.apache.ignite.testframework.GridTestUtils;
-
-import java.net.URI;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Map;
-import java.util.UUID;
-
-/**
- *
- */
-public class HadoopDefaultMapReducePlannerSelfTest extends HadoopAbstractSelfTest {
- /** */
- private static final UUID ID_1 = new UUID(0, 1);
-
- /** */
- private static final UUID ID_2 = new UUID(0, 2);
-
- /** */
- private static final UUID ID_3 = new UUID(0, 3);
-
- /** */
- private static final String HOST_1 = "host1";
-
- /** */
- private static final String HOST_2 = "host2";
-
- /** */
- private static final String HOST_3 = "host3";
-
- /** */
- private static final String INVALID_HOST_1 = "invalid_host1";
-
- /** */
- private static final String INVALID_HOST_2 = "invalid_host2";
-
- /** */
- private static final String INVALID_HOST_3 = "invalid_host3";
-
- /** Mocked IGFS. */
- private static final IgniteFileSystem IGFS = new MockIgfs();
-
- /** Mocked Grid. */
- private static final IgfsIgniteMock GRID = new IgfsIgniteMock(null, IGFS);
-
- /** Planner. */
- private static final HadoopMapReducePlanner PLANNER = new IgniteHadoopMapReducePlanner();
-
- /** Block locations. */
- private static final Map<Block, Collection<IgfsBlockLocation>> BLOCK_MAP = new HashMap<>();
-
- /** Proxy map. */
- private static final Map<URI, Boolean> PROXY_MAP = new HashMap<>();
-
- /** Last created plan. */
- private static final ThreadLocal<HadoopMapReducePlan> PLAN = new ThreadLocal<>();
-
- /**
- * Static initializer.
- */
- static {
- GridTestUtils.setFieldValue(PLANNER, HadoopAbstractMapReducePlanner.class, "ignite", GRID);
- }
-
- /** {@inheritDoc} */
- @Override protected void beforeTest() throws Exception {
- GridTestUtils.setFieldValue(PLANNER, HadoopAbstractMapReducePlanner.class, "log", log());
-
- BLOCK_MAP.clear();
- PROXY_MAP.clear();
- }
-
- /**
- * @throws IgniteCheckedException If failed.
- */
- public void testIgfsOneBlockPerNode() throws IgniteCheckedException {
- HadoopFileBlock split1 = split(true, "/file1", 0, 100, HOST_1);
- HadoopFileBlock split2 = split(true, "/file2", 0, 100, HOST_2);
- HadoopFileBlock split3 = split(true, "/file3", 0, 100, HOST_3);
-
- mapIgfsBlock(split1.file(), 0, 100, location(0, 100, ID_1));
- mapIgfsBlock(split2.file(), 0, 100, location(0, 100, ID_2));
- mapIgfsBlock(split3.file(), 0, 100, location(0, 100, ID_3));
-
- plan(1, split1);
- assert ensureMappers(ID_1, split1);
- assert ensureReducers(ID_1, 1);
- assert ensureEmpty(ID_2);
- assert ensureEmpty(ID_3);
-
- plan(2, split1);
- assert ensureMappers(ID_1, split1);
- assert ensureReducers(ID_1, 2);
- assert ensureEmpty(ID_2);
- assert ensureEmpty(ID_3);
-
- plan(1, split1, split2);
- assert ensureMappers(ID_1, split1);
- assert ensureMappers(ID_2, split2);
- assert ensureReducers(ID_1, 1) && ensureReducers(ID_2, 0) || ensureReducers(ID_1, 0) && ensureReducers(ID_2, 1);
- assert ensureEmpty(ID_3);
-
- plan(2, split1, split2);
- assert ensureMappers(ID_1, split1);
- assert ensureMappers(ID_2, split2);
- assert ensureReducers(ID_1, 1);
- assert ensureReducers(ID_2, 1);
- assert ensureEmpty(ID_3);
-
- plan(3, split1, split2);
- assert ensureMappers(ID_1, split1);
- assert ensureMappers(ID_2, split2);
- assert ensureReducers(ID_1, 1) && ensureReducers(ID_2, 2) || ensureReducers(ID_1, 2) && ensureReducers(ID_2, 1);
- assert ensureEmpty(ID_3);
-
- plan(3, split1, split2, split3);
- assert ensureMappers(ID_1, split1);
- assert ensureMappers(ID_2, split2);
- assert ensureMappers(ID_3, split3);
- assert ensureReducers(ID_1, 1);
- assert ensureReducers(ID_2, 1);
- assert ensureReducers(ID_3, 1);
-
- plan(5, split1, split2, split3);
- assert ensureMappers(ID_1, split1);
- assert ensureMappers(ID_2, split2);
- assert ensureMappers(ID_3, split3);
- assert ensureReducers(ID_1, 1) && ensureReducers(ID_2, 2) && ensureReducers(ID_3, 2) ||
- ensureReducers(ID_1, 2) && ensureReducers(ID_2, 1) && ensureReducers(ID_3, 2) ||
- ensureReducers(ID_1, 2) && ensureReducers(ID_2, 2) && ensureReducers(ID_3, 1);
- }
-
- /**
- * @throws IgniteCheckedException If failed.
- */
- public void testNonIgfsOneBlockPerNode() throws IgniteCheckedException {
- HadoopFileBlock split1 = split(false, "/file1", 0, 100, HOST_1);
- HadoopFileBlock split2 = split(false, "/file2", 0, 100, HOST_2);
- HadoopFileBlock split3 = split(false, "/file3", 0, 100, HOST_3);
-
- plan(1, split1);
- assert ensureMappers(ID_1, split1);
- assert ensureReducers(ID_1, 1);
- assert ensureEmpty(ID_2);
- assert ensureEmpty(ID_3);
-
- plan(2, split1);
- assert ensureMappers(ID_1, split1);
- assert ensureReducers(ID_1, 2);
- assert ensureEmpty(ID_2);
- assert ensureEmpty(ID_3);
-
- plan(1, split1, split2);
- assert ensureMappers(ID_1, split1);
- assert ensureMappers(ID_2, split2);
- assert ensureReducers(ID_1, 1) && ensureReducers(ID_2, 0) || ensureReducers(ID_1, 0) && ensureReducers(ID_2, 1);
- assert ensureEmpty(ID_3);
-
- plan(2, split1, split2);
- assert ensureMappers(ID_1, split1);
- assert ensureMappers(ID_2, split2);
- assert ensureReducers(ID_1, 1);
- assert ensureReducers(ID_2, 1);
- assert ensureEmpty(ID_3);
-
- plan(3, split1, split2);
- assert ensureMappers(ID_1, split1);
- assert ensureMappers(ID_2, split2);
- assert ensureReducers(ID_1, 1) && ensureReducers(ID_2, 2) || ensureReducers(ID_1, 2) && ensureReducers(ID_2, 1);
- assert ensureEmpty(ID_3);
-
- plan(3, split1, split2, split3);
- assert ensureMappers(ID_1, split1);
- assert ensureMappers(ID_2, split2);
- assert ensureMappers(ID_3, split3);
- assert ensureReducers(ID_1, 1);
- assert ensureReducers(ID_2, 1);
- assert ensureReducers(ID_3, 1);
-
- plan(5, split1, split2, split3);
- assert ensureMappers(ID_1, split1);
- assert ensureMappers(ID_2, split2);
- assert ensureMappers(ID_3, split3);
- assert ensureReducers(ID_1, 1) && ensureReducers(ID_2, 2) && ensureReducers(ID_3, 2) ||
- ensureReducers(ID_1, 2) && ensureReducers(ID_2, 1) && ensureReducers(ID_3, 2) ||
- ensureReducers(ID_1, 2) && ensureReducers(ID_2, 2) && ensureReducers(ID_3, 1);
- }
-
- /**
- * @throws IgniteCheckedException If failed.
- */
- public void testIgfsSeveralBlocksPerNode() throws IgniteCheckedException {
- HadoopFileBlock split1 = split(true, "/file1", 0, 100, HOST_1, HOST_2);
- HadoopFileBlock split2 = split(true, "/file2", 0, 100, HOST_1, HOST_2);
- HadoopFileBlock split3 = split(true, "/file3", 0, 100, HOST_1, HOST_3);
-
- mapIgfsBlock(split1.file(), 0, 100, location(0, 100, ID_1, ID_2));
- mapIgfsBlock(split2.file(), 0, 100, location(0, 100, ID_1, ID_2));
- mapIgfsBlock(split3.file(), 0, 100, location(0, 100, ID_1, ID_3));
-
- plan(1, split1);
- assert ensureMappers(ID_1, split1) && ensureReducers(ID_1, 1) && ensureEmpty(ID_2) ||
- ensureEmpty(ID_1) && ensureMappers(ID_2, split1) && ensureReducers(ID_2, 1);
- assert ensureEmpty(ID_3);
-
- plan(2, split1);
- assert ensureMappers(ID_1, split1) && ensureReducers(ID_1, 2) && ensureEmpty(ID_2) ||
- ensureEmpty(ID_1) && ensureMappers(ID_2, split1) && ensureReducers(ID_2, 2);
- assert ensureEmpty(ID_3);
-
- plan(1, split1, split2);
- assert ensureMappers(ID_1, split1) && ensureMappers(ID_2, split2) ||
- ensureMappers(ID_1, split2) && ensureMappers(ID_2, split1);
- assert ensureReducers(ID_1, 1) && ensureReducers(ID_2, 0) || ensureReducers(ID_1, 0) && ensureReducers(ID_2, 1);
- assert ensureEmpty(ID_3);
-
- plan(2, split1, split2);
- assert ensureMappers(ID_1, split1) && ensureMappers(ID_2, split2) ||
- ensureMappers(ID_1, split2) && ensureMappers(ID_2, split1);
- assert ensureReducers(ID_1, 1);
- assert ensureReducers(ID_2, 1);
- assert ensureEmpty(ID_3);
-
- plan(3, split1, split2, split3);
- assert ensureReducers(ID_1, 1);
- assert ensureReducers(ID_2, 1);
- assert ensureReducers(ID_3, 1);
-
- plan(5, split1, split2, split3);
- assert ensureReducers(ID_1, 1) && ensureReducers(ID_2, 2) && ensureReducers(ID_3, 2) ||
- ensureReducers(ID_1, 2) && ensureReducers(ID_2, 1) && ensureReducers(ID_3, 2) ||
- ensureReducers(ID_1, 2) && ensureReducers(ID_2, 2) && ensureReducers(ID_3, 1);
- }
-
- /**
- * @throws IgniteCheckedException If failed.
- */
- public void testNonIgfsSeveralBlocksPerNode() throws IgniteCheckedException {
- HadoopFileBlock split1 = split(false, "/file1", 0, 100, HOST_1, HOST_2);
- HadoopFileBlock split2 = split(false, "/file2", 0, 100, HOST_1, HOST_2);
- HadoopFileBlock split3 = split(false, "/file3", 0, 100, HOST_1, HOST_3);
-
- plan(1, split1);
- assert ensureMappers(ID_1, split1) && ensureReducers(ID_1, 1) && ensureEmpty(ID_2) ||
- ensureEmpty(ID_1) && ensureMappers(ID_2, split1) && ensureReducers(ID_2, 1);
- assert ensureEmpty(ID_3);
-
- plan(2, split1);
- assert ensureMappers(ID_1, split1) && ensureReducers(ID_1, 2) && ensureEmpty(ID_2) ||
- ensureEmpty(ID_1) && ensureMappers(ID_2, split1) && ensureReducers(ID_2, 2);
- assert ensureEmpty(ID_3);
-
- plan(1, split1, split2);
- assert ensureMappers(ID_1, split1) && ensureMappers(ID_2, split2) ||
- ensureMappers(ID_1, split2) && ensureMappers(ID_2, split1);
- assert ensureReducers(ID_1, 1) && ensureReducers(ID_2, 0) || ensureReducers(ID_1, 0) && ensureReducers(ID_2, 1);
- assert ensureEmpty(ID_3);
-
- plan(2, split1, split2);
- assert ensureMappers(ID_1, split1) && ensureMappers(ID_2, split2) ||
- ensureMappers(ID_1, split2) && ensureMappers(ID_2, split1);
- assert ensureReducers(ID_1, 1);
- assert ensureReducers(ID_2, 1);
- assert ensureEmpty(ID_3);
-
- plan(3, split1, split2, split3);
- assert ensureReducers(ID_1, 1);
- assert ensureReducers(ID_2, 1);
- assert ensureReducers(ID_3, 1);
-
- plan(5, split1, split2, split3);
- assert ensureReducers(ID_1, 1) && ensureReducers(ID_2, 2) && ensureReducers(ID_3, 2) ||
- ensureReducers(ID_1, 2) && ensureReducers(ID_2, 1) && ensureReducers(ID_3, 2) ||
- ensureReducers(ID_1, 2) && ensureReducers(ID_2, 2) && ensureReducers(ID_3, 1);
- }
-
- /**
- * @throws IgniteCheckedException If failed.
- */
- public void testIgfsSeveralComplexBlocksPerNode() throws IgniteCheckedException {
- HadoopFileBlock split1 = split(true, "/file1", 0, 100, HOST_1, HOST_2, HOST_3);
- HadoopFileBlock split2 = split(true, "/file2", 0, 100, HOST_1, HOST_2, HOST_3);
-
- mapIgfsBlock(split1.file(), 0, 100, location(0, 50, ID_1, ID_2), location(51, 100, ID_1, ID_3));
- mapIgfsBlock(split2.file(), 0, 100, location(0, 50, ID_1, ID_2), location(51, 100, ID_2, ID_3));
-
- plan(1, split1);
- assert ensureMappers(ID_1, split1);
- assert ensureReducers(ID_1, 1);
- assert ensureEmpty(ID_2);
- assert ensureEmpty(ID_3);
-
- plan(1, split2);
- assert ensureMappers(ID_2, split2);
- assert ensureReducers(ID_2, 1);
- assert ensureEmpty(ID_1);
- assert ensureEmpty(ID_3);
-
- plan(1, split1, split2);
- assert ensureMappers(ID_1, split1);
- assert ensureMappers(ID_2, split2);
- assert ensureReducers(ID_1, 0) && ensureReducers(ID_2, 1) || ensureReducers(ID_1, 1) && ensureReducers(ID_2, 0);
- assert ensureEmpty(ID_3);
-
- plan(2, split1, split2);
- assert ensureMappers(ID_1, split1);
- assert ensureMappers(ID_2, split2);
- assert ensureReducers(ID_1, 1);
- assert ensureReducers(ID_2, 1);
- assert ensureEmpty(ID_3);
- }
-
- /**
- * @throws IgniteCheckedException If failed.
- */
- public void testNonIgfsOrphans() throws IgniteCheckedException {
- HadoopFileBlock split1 = split(false, "/file1", 0, 100, INVALID_HOST_1, INVALID_HOST_2);
- HadoopFileBlock split2 = split(false, "/file2", 0, 100, INVALID_HOST_1, INVALID_HOST_3);
- HadoopFileBlock split3 = split(false, "/file3", 0, 100, INVALID_HOST_2, INVALID_HOST_3);
-
- plan(1, split1);
- assert ensureMappers(ID_1, split1) && ensureReducers(ID_1, 1) && ensureEmpty(ID_2) && ensureEmpty(ID_3) ||
- ensureEmpty(ID_1) && ensureMappers(ID_2, split1) && ensureReducers(ID_2, 1) && ensureEmpty(ID_3) ||
- ensureEmpty(ID_1) && ensureEmpty(ID_2) && ensureMappers(ID_3, split1) && ensureReducers(ID_3, 1);
-
- plan(2, split1);
- assert ensureMappers(ID_1, split1) && ensureReducers(ID_1, 2) && ensureEmpty(ID_2) && ensureEmpty(ID_3) ||
- ensureEmpty(ID_1) && ensureMappers(ID_2, split1) && ensureReducers(ID_2, 2) && ensureEmpty(ID_3) ||
- ensureEmpty(ID_1) && ensureEmpty(ID_2) && ensureMappers(ID_3, split1) && ensureReducers(ID_3, 2);
-
- plan(1, split1, split2, split3);
- assert ensureMappers(ID_1, split1) && ensureMappers(ID_2, split2) && ensureMappers(ID_3, split3) ||
- ensureMappers(ID_1, split1) && ensureMappers(ID_2, split3) && ensureMappers(ID_3, split2) ||
- ensureMappers(ID_1, split2) && ensureMappers(ID_2, split1) && ensureMappers(ID_3, split3) ||
- ensureMappers(ID_1, split2) && ensureMappers(ID_2, split3) && ensureMappers(ID_3, split1) ||
- ensureMappers(ID_1, split3) && ensureMappers(ID_2, split1) && ensureMappers(ID_3, split2) ||
- ensureMappers(ID_1, split3) && ensureMappers(ID_2, split2) && ensureMappers(ID_3, split1);
- assert ensureReducers(ID_1, 1) && ensureReducers(ID_2, 0) && ensureReducers(ID_3, 0) ||
- ensureReducers(ID_1, 0) && ensureReducers(ID_2, 1) && ensureReducers(ID_3, 0) ||
- ensureReducers(ID_1, 0) && ensureReducers(ID_2, 0) && ensureReducers(ID_3, 1);
-
- plan(3, split1, split2, split3);
- assert ensureMappers(ID_1, split1) && ensureMappers(ID_2, split2) && ensureMappers(ID_3, split3) ||
- ensureMappers(ID_1, split1) && ensureMappers(ID_2, split3) && ensureMappers(ID_3, split2) ||
- ensureMappers(ID_1, split2) && ensureMappers(ID_2, split1) && ensureMappers(ID_3, split3) ||
- ensureMappers(ID_1, split2) && ensureMappers(ID_2, split3) && ensureMappers(ID_3, split1) ||
- ensureMappers(ID_1, split3) && ensureMappers(ID_2, split1) && ensureMappers(ID_3, split2) ||
- ensureMappers(ID_1, split3) && ensureMappers(ID_2, split2) && ensureMappers(ID_3, split1);
- assert ensureReducers(ID_1, 1);
- assert ensureReducers(ID_2, 1);
- assert ensureReducers(ID_3, 1);
-
- plan(5, split1, split2, split3);
- assert ensureMappers(ID_1, split1) && ensureMappers(ID_2, split2) && ensureMappers(ID_3, split3) ||
- ensureMappers(ID_1, split1) && ensureMappers(ID_2, split3) && ensureMappers(ID_3, split2) ||
- ensureMappers(ID_1, split2) && ensureMappers(ID_2, split1) && ensureMappers(ID_3, split3) ||
- ensureMappers(ID_1, split2) && ensureMappers(ID_2, split3) && ensureMappers(ID_3, split1) ||
- ensureMappers(ID_1, split3) && ensureMappers(ID_2, split1) && ensureMappers(ID_3, split2) ||
- ensureMappers(ID_1, split3) && ensureMappers(ID_2, split2) && ensureMappers(ID_3, split1);
- assert ensureReducers(ID_1, 1) && ensureReducers(ID_2, 2) && ensureReducers(ID_3, 2) ||
- ensureReducers(ID_1, 2) && ensureReducers(ID_2, 1) && ensureReducers(ID_3, 2) ||
- ensureReducers(ID_1, 2) && ensureReducers(ID_2, 2) && ensureReducers(ID_3, 1);
- }
-
- /**
- * Create plan.
- *
- * @param reducers Reducers count.
- * @param splits Splits.
- * @return Plan.
- * @throws IgniteCheckedException If failed.
- */
- private static HadoopMapReducePlan plan(int reducers, HadoopInputSplit... splits) throws IgniteCheckedException {
- assert reducers > 0;
- assert splits != null && splits.length > 0;
-
- Collection<HadoopInputSplit> splitList = new ArrayList<>(splits.length);
-
- Collections.addAll(splitList, splits);
-
- Collection<ClusterNode> top = new ArrayList<>();
-
- GridTestNode node1 = new GridTestNode(ID_1);
- GridTestNode node2 = new GridTestNode(ID_2);
- GridTestNode node3 = new GridTestNode(ID_3);
-
- node1.setHostName(HOST_1);
- node2.setHostName(HOST_2);
- node3.setHostName(HOST_3);
-
- top.add(node1);
- top.add(node2);
- top.add(node3);
-
- HadoopMapReducePlan plan = PLANNER.preparePlan(new HadoopPlannerMockJob(splitList, reducers), top, null);
-
- PLAN.set(plan);
-
- return plan;
- }
-
- /**
- * Ensure that node contains the given mappers.
- *
- * @param nodeId Node ID.
- * @param expSplits Expected splits.
- * @return {@code True} if this assumption is valid.
- */
- private static boolean ensureMappers(UUID nodeId, HadoopInputSplit... expSplits) {
- Collection<HadoopInputSplit> expSplitsCol = new ArrayList<>();
-
- Collections.addAll(expSplitsCol, expSplits);
-
- Collection<HadoopInputSplit> splits = PLAN.get().mappers(nodeId);
-
- return F.eq(expSplitsCol, splits);
- }
-
- /**
- * Ensure that node contains the given amount of reducers.
- *
- * @param nodeId Node ID.
- * @param reducers Reducers.
- * @return {@code True} if this assumption is valid.
- */
- private static boolean ensureReducers(UUID nodeId, int reducers) {
- int[] reducersArr = PLAN.get().reducers(nodeId);
-
- return reducers == 0 ? F.isEmpty(reducersArr) : (reducersArr != null && reducersArr.length == reducers);
- }
-
- /**
- * Ensure that no mappers and reducers is located on this node.
- *
- * @param nodeId Node ID.
- * @return {@code True} if this assumption is valid.
- */
- private static boolean ensureEmpty(UUID nodeId) {
- return F.isEmpty(PLAN.get().mappers(nodeId)) && F.isEmpty(PLAN.get().reducers(nodeId));
- }
-
- /**
- * Create split.
- *
- * @param igfs IGFS flag.
- * @param file File.
- * @param start Start.
- * @param len Length.
- * @param hosts Hosts.
- * @return Split.
- */
- private static HadoopFileBlock split(boolean igfs, String file, long start, long len, String... hosts) {
- URI uri = URI.create((igfs ? "igfs://igfs@" : "hdfs://") + file);
-
- return new HadoopFileBlock(hosts, uri, start, len);
- }
-
- /**
- * Create block location.
- *
- * @param start Start.
- * @param len Length.
- * @param nodeIds Node IDs.
- * @return Block location.
- */
- private static IgfsBlockLocation location(long start, long len, UUID... nodeIds) {
- assert nodeIds != null && nodeIds.length > 0;
-
- Collection<ClusterNode> nodes = new ArrayList<>(nodeIds.length);
-
- for (UUID id : nodeIds)
- nodes.add(new GridTestNode(id));
-
- return new IgfsBlockLocationImpl(start, len, nodes);
- }
-
- /**
- * Map IGFS block to nodes.
- *
- * @param file File.
- * @param start Start.
- * @param len Length.
- * @param locations Locations.
- */
- private static void mapIgfsBlock(URI file, long start, long len, IgfsBlockLocation... locations) {
- assert locations != null && locations.length > 0;
-
- IgfsPath path = new IgfsPath(file);
-
- Block block = new Block(path, start, len);
-
- Collection<IgfsBlockLocation> locationsList = new ArrayList<>();
-
- Collections.addAll(locationsList, locations);
-
- BLOCK_MAP.put(block, locationsList);
- }
-
- /**
- * Block.
- */
- private static class Block {
- /** */
- private final IgfsPath path;
-
- /** */
- private final long start;
-
- /** */
- private final long len;
-
- /**
- * Constructor.
- *
- * @param path Path.
- * @param start Start.
- * @param len Length.
- */
- private Block(IgfsPath path, long start, long len) {
- this.path = path;
- this.start = start;
- this.len = len;
- }
-
- /** {@inheritDoc} */
- @SuppressWarnings("RedundantIfStatement")
- @Override public boolean equals(Object o) {
- if (this == o) return true;
- if (!(o instanceof Block)) return false;
-
- Block block = (Block) o;
-
- if (len != block.len)
- return false;
-
- if (start != block.start)
- return false;
-
- if (!path.equals(block.path))
- return false;
-
- return true;
- }
-
- /** {@inheritDoc} */
- @Override public int hashCode() {
- int res = path.hashCode();
-
- res = 31 * res + (int) (start ^ (start >>> 32));
- res = 31 * res + (int) (len ^ (len >>> 32));
-
- return res;
- }
- }
-
- /**
- * Mocked IGFS.
- */
- private static class MockIgfs extends IgfsMock {
- /**
- * Constructor.
- */
- public MockIgfs() {
- super("igfs");
- }
-
- /** {@inheritDoc} */
- @Override public boolean isProxy(URI path) {
- return PROXY_MAP.containsKey(path) && PROXY_MAP.get(path);
- }
-
- /** {@inheritDoc} */
- @Override public Collection<IgfsBlockLocation> affinity(IgfsPath path, long start, long len) {
- return BLOCK_MAP.get(new Block(path, start, len));
- }
-
- /** {@inheritDoc} */
- @Override public boolean exists(IgfsPath path) {
- return true;
- }
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/8032fc2c/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopErrorSimulator.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopErrorSimulator.java b/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopErrorSimulator.java
deleted file mode 100644
index 843b42b..0000000
--- a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopErrorSimulator.java
+++ /dev/null
@@ -1,326 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.processors.hadoop;
-
-import java.io.IOException;
-import java.util.concurrent.atomic.AtomicReference;
-
-/**
- * Error simulator.
- */
-public class HadoopErrorSimulator {
- /** No-op singleton instance. */
- public static final HadoopErrorSimulator noopInstance = new HadoopErrorSimulator();
-
- /** Instance ref. */
- private static final AtomicReference<HadoopErrorSimulator> ref = new AtomicReference<>(noopInstance);
-
- /**
- * Creates simulator of given kind with given stage bits.
- *
- * @param kind The kind.
- * @param bits The stage bits.
- * @return The simulator.
- */
- public static HadoopErrorSimulator create(Kind kind, int bits) {
- switch (kind) {
- case Noop:
- return noopInstance;
- case Runtime:
- return new RuntimeExceptionBitHadoopErrorSimulator(bits);
- case IOException:
- return new IOExceptionBitHadoopErrorSimulator(bits);
- case Error:
- return new ErrorBitHadoopErrorSimulator(bits);
- default:
- throw new IllegalStateException("Unknown kind: " + kind);
- }
- }
-
- /**
- * Gets the error simulator instance.
- */
- public static HadoopErrorSimulator instance() {
- return ref.get();
- }
-
- /**
- * Sets instance.
- */
- public static boolean setInstance(HadoopErrorSimulator expect, HadoopErrorSimulator update) {
- return ref.compareAndSet(expect, update);
- }
-
- /**
- * Constructor.
- */
- private HadoopErrorSimulator() {
- // no-op
- }
-
- /**
- * Invoked on the named stage.
- */
- public void onMapConfigure() {
- // no-op
- }
-
- /**
- * Invoked on the named stage.
- */
- public void onMapSetup() throws IOException, InterruptedException {
- // no-op
- }
-
- /**
- * Invoked on the named stage.
- */
- public void onMap() throws IOException {
- // no-op
- }
-
- /**
- * Invoked on the named stage.
- */
- public void onMapCleanup() throws IOException, InterruptedException {
- // no-op
- }
-
- /**
- * Invoked on the named stage.
- */
- public void onMapClose() throws IOException {
- // no-op
- }
-
- /**
- * setConf() does not declare IOException to be thrown.
- */
- public void onCombineConfigure() {
- // no-op
- }
-
- /**
- * Invoked on the named stage.
- */
- public void onCombineSetup() throws IOException, InterruptedException {
- // no-op
- }
-
- /**
- * Invoked on the named stage.
- */
- public void onCombine() throws IOException {
- // no-op
- }
-
- /**
- * Invoked on the named stage.
- */
- public void onCombineCleanup() throws IOException, InterruptedException {
- // no-op
- }
-
- /**
- * Invoked on the named stage.
- */
- public void onReduceConfigure() {
- // no-op
- }
-
- /**
- * Invoked on the named stage.
- */
- public void onReduceSetup() throws IOException, InterruptedException {
- // no-op
- }
-
- /**
- * Invoked on the named stage.
- */
- public void onReduce() throws IOException {
- // no-op
- }
-
- /**
- * Invoked on the named stage.
- */
- public void onReduceCleanup() throws IOException, InterruptedException {
- // no-op
- }
-
- /**
- * Error kind.
- */
- public enum Kind {
- /** No error. */
- Noop,
-
- /** Runtime. */
- Runtime,
-
- /** IOException. */
- IOException,
-
- /** java.lang.Error. */
- Error
- }
-
- /**
- * Runtime error simulator.
- */
- public static class RuntimeExceptionBitHadoopErrorSimulator extends HadoopErrorSimulator {
- /** Stage bits: defines what map-reduce stages will cause errors. */
- private final int bits;
-
- /**
- * Constructor.
- */
- protected RuntimeExceptionBitHadoopErrorSimulator(int b) {
- bits = b;
- }
-
- /**
- * Simulates an error.
- */
- protected void simulateError() throws IOException {
- throw new RuntimeException("An error simulated by " + getClass().getSimpleName());
- }
-
- /** {@inheritDoc} */
- @Override public final void onMapConfigure() {
- try {
- if ((bits & 1) != 0)
- simulateError();
- }
- catch (IOException e) {
- // ignore
- }
- }
-
- /** {@inheritDoc} */
- @Override public final void onMapSetup() throws IOException, InterruptedException {
- if ((bits & 2) != 0)
- simulateError();
- }
-
- /** {@inheritDoc} */
- @Override public final void onMap() throws IOException {
- if ((bits & 4) != 0)
- simulateError();
- }
-
- /** {@inheritDoc} */
- @Override public final void onMapCleanup() throws IOException, InterruptedException {
- if ((bits & 8) != 0)
- simulateError();
- }
-
- /** {@inheritDoc} */
- @Override public final void onCombineConfigure() {
- try {
- if ((bits & 16) != 0)
- simulateError();
- }
- catch (IOException e) {
- // ignore
- }
- }
-
- /** {@inheritDoc} */
- @Override public final void onCombineSetup() throws IOException, InterruptedException {
- if ((bits & 32) != 0)
- simulateError();
- }
-
- /** {@inheritDoc} */
- @Override public final void onCombine() throws IOException {
- if ((bits & 64) != 0)
- simulateError();
- }
-
- /** {@inheritDoc} */
- @Override public final void onCombineCleanup() throws IOException, InterruptedException {
- if ((bits & 128) != 0)
- simulateError();
- }
-
- /** {@inheritDoc} */
- @Override public final void onReduceConfigure() {
- try {
- if ((bits & 256) != 0)
- simulateError();
- }
- catch (IOException e) {
- // ignore
- }
- }
-
- /** {@inheritDoc} */
- @Override public final void onReduceSetup() throws IOException, InterruptedException {
- if ((bits & 512) != 0)
- simulateError();
- }
-
- /** {@inheritDoc} */
- @Override public final void onReduce() throws IOException {
- if ((bits & 1024) != 0)
- simulateError();
- }
-
- /** {@inheritDoc} */
- @Override public final void onReduceCleanup() throws IOException, InterruptedException {
- if ((bits & 2048) != 0)
- simulateError();
- }
- }
-
- /**
- * java.lang.Error simulator.
- */
- public static class ErrorBitHadoopErrorSimulator extends RuntimeExceptionBitHadoopErrorSimulator {
- /**
- * Constructor.
- */
- public ErrorBitHadoopErrorSimulator(int bits) {
- super(bits);
- }
-
- /** {@inheritDoc} */
- @Override protected void simulateError() {
- throw new Error("An error simulated by " + getClass().getSimpleName());
- }
- }
-
- /**
- * IOException simulator.
- */
- public static class IOExceptionBitHadoopErrorSimulator extends RuntimeExceptionBitHadoopErrorSimulator {
- /**
- * Constructor.
- */
- public IOExceptionBitHadoopErrorSimulator(int bits) {
- super(bits);
- }
-
- /** {@inheritDoc} */
- @Override protected void simulateError() throws IOException {
- throw new IOException("An IOException simulated by " + getClass().getSimpleName());
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/ignite/blob/8032fc2c/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopFileSystemsTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopFileSystemsTest.java b/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopFileSystemsTest.java
deleted file mode 100644
index 946ba77..0000000
--- a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopFileSystemsTest.java
+++ /dev/null
@@ -1,155 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.processors.hadoop;
-
-import java.io.IOException;
-import java.net.URI;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.atomic.AtomicInteger;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.mapreduce.MRJobConfig;
-import org.apache.ignite.internal.processors.hadoop.fs.HadoopFileSystemsUtils;
-import org.apache.ignite.testframework.GridTestUtils;
-
-/**
- * Test file systems for the working directory multi-threading support.
- */
-public class HadoopFileSystemsTest extends HadoopAbstractSelfTest {
- /** the number of threads */
- private static final int THREAD_COUNT = 3;
-
- /** {@inheritDoc} */
- @Override protected void beforeTest() throws Exception {
- startGrids(gridCount());
- }
-
- /** {@inheritDoc} */
- @Override protected void afterTest() throws Exception {
- stopAllGrids(true);
- }
-
- /** {@inheritDoc} */
- @Override protected boolean igfsEnabled() {
- return true;
- }
-
- /** {@inheritDoc} */
- @Override protected int gridCount() {
- return 1;
- }
-
-
- /**
- * Test the file system with specified URI for the multi-thread working directory support.
- *
- * @param uri Base URI of the file system (scheme and authority).
- * @throws Exception If fails.
- */
- private void testFileSystem(final URI uri) throws Exception {
- final Configuration cfg = new Configuration();
-
- setupFileSystems(cfg);
-
- cfg.set(HadoopFileSystemsUtils.LOC_FS_WORK_DIR_PROP,
- new Path(new Path(uri), "user/" + System.getProperty("user.name")).toString());
-
- final CountDownLatch changeUserPhase = new CountDownLatch(THREAD_COUNT);
- final CountDownLatch changeDirPhase = new CountDownLatch(THREAD_COUNT);
- final CountDownLatch changeAbsDirPhase = new CountDownLatch(THREAD_COUNT);
- final CountDownLatch finishPhase = new CountDownLatch(THREAD_COUNT);
-
- final Path[] newUserInitWorkDir = new Path[THREAD_COUNT];
- final Path[] newWorkDir = new Path[THREAD_COUNT];
- final Path[] newAbsWorkDir = new Path[THREAD_COUNT];
- final Path[] newInstanceWorkDir = new Path[THREAD_COUNT];
-
- final AtomicInteger threadNum = new AtomicInteger(0);
-
- GridTestUtils.runMultiThreadedAsync(new Runnable() {
- @Override public void run() {
- try {
- int curThreadNum = threadNum.getAndIncrement();
-
- if ("file".equals(uri.getScheme()))
- FileSystem.get(uri, cfg).setWorkingDirectory(new Path("file:///user/user" + curThreadNum));
-
- changeUserPhase.countDown();
- changeUserPhase.await();
-
- newUserInitWorkDir[curThreadNum] = FileSystem.get(uri, cfg).getWorkingDirectory();
-
- FileSystem.get(uri, cfg).setWorkingDirectory(new Path("folder" + curThreadNum));
-
- changeDirPhase.countDown();
- changeDirPhase.await();
-
- newWorkDir[curThreadNum] = FileSystem.get(uri, cfg).getWorkingDirectory();
-
- FileSystem.get(uri, cfg).setWorkingDirectory(new Path("/folder" + curThreadNum));
-
- changeAbsDirPhase.countDown();
- changeAbsDirPhase.await();
-
- newAbsWorkDir[curThreadNum] = FileSystem.get(uri, cfg).getWorkingDirectory();
-
- newInstanceWorkDir[curThreadNum] = FileSystem.newInstance(uri, cfg).getWorkingDirectory();
-
- finishPhase.countDown();
- }
- catch (InterruptedException | IOException e) {
- error("Failed to execute test thread.", e);
-
- fail();
- }
- }
- }, THREAD_COUNT, "filesystems-test");
-
- finishPhase.await();
-
- for (int i = 0; i < THREAD_COUNT; i ++) {
- cfg.set(MRJobConfig.USER_NAME, "user" + i);
-
- Path workDir = new Path(new Path(uri), "user/user" + i);
-
- cfg.set(HadoopFileSystemsUtils.LOC_FS_WORK_DIR_PROP, workDir.toString());
-
- assertEquals(workDir, FileSystem.newInstance(uri, cfg).getWorkingDirectory());
-
- assertEquals(workDir, newUserInitWorkDir[i]);
-
- assertEquals(new Path(new Path(uri), "user/user" + i + "/folder" + i), newWorkDir[i]);
-
- assertEquals(new Path("/folder" + i), newAbsWorkDir[i]);
-
- assertEquals(new Path(new Path(uri), "user/" + System.getProperty("user.name")), newInstanceWorkDir[i]);
- }
-
- System.out.println(System.getProperty("user.dir"));
- }
-
- /**
- * Test LocalFS multi-thread working directory.
- *
- * @throws Exception If fails.
- */
- public void testLocal() throws Exception {
- testFileSystem(URI.create("file:///"));
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/8032fc2c/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopGroupingTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopGroupingTest.java b/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopGroupingTest.java
deleted file mode 100644
index db87e33..0000000
--- a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopGroupingTest.java
+++ /dev/null
@@ -1,307 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.processors.hadoop;
-
-import java.io.DataInput;
-import java.io.DataOutput;
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Random;
-import java.util.Set;
-import java.util.UUID;
-import org.apache.hadoop.io.RawComparator;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.io.WritableComparable;
-import org.apache.hadoop.mapreduce.InputFormat;
-import org.apache.hadoop.mapreduce.InputSplit;
-import org.apache.hadoop.mapreduce.Job;
-import org.apache.hadoop.mapreduce.JobContext;
-import org.apache.hadoop.mapreduce.Mapper;
-import org.apache.hadoop.mapreduce.OutputCommitter;
-import org.apache.hadoop.mapreduce.OutputFormat;
-import org.apache.hadoop.mapreduce.RecordReader;
-import org.apache.hadoop.mapreduce.RecordWriter;
-import org.apache.hadoop.mapreduce.Reducer;
-import org.apache.hadoop.mapreduce.TaskAttemptContext;
-import org.apache.ignite.configuration.HadoopConfiguration;
-import org.apache.ignite.internal.util.GridConcurrentHashSet;
-import org.apache.ignite.internal.util.GridRandom;
-import org.apache.ignite.internal.util.typedef.X;
-import org.apache.ignite.internal.util.typedef.internal.S;
-
-import static org.apache.ignite.internal.processors.hadoop.HadoopUtils.createJobInfo;
-
-/**
- * Grouping test.
- */
-public class HadoopGroupingTest extends HadoopAbstractSelfTest {
- /** */
- private static final String PATH_OUTPUT = "/test-out";
-
- /** */
- private static final GridConcurrentHashSet<UUID> vals = HadoopSharedMap.map(HadoopGroupingTest.class)
- .put("vals", new GridConcurrentHashSet<UUID>());
-
- /** {@inheritDoc} */
- @Override protected int gridCount() {
- return 3;
- }
-
- /** {@inheritDoc} */
- protected boolean igfsEnabled() {
- return false;
- }
-
- /** {@inheritDoc} */
- @Override protected void beforeTest() throws Exception {
- startGrids(gridCount());
- }
-
- /** {@inheritDoc} */
- @Override protected void afterTest() throws Exception {
- stopAllGrids(true);
- }
-
- /** {@inheritDoc} */
- @Override public HadoopConfiguration hadoopConfiguration(String gridName) {
- HadoopConfiguration cfg = super.hadoopConfiguration(gridName);
-
- // TODO: IGNITE-404: Uncomment when fixed.
- //cfg.setExternalExecution(false);
-
- return cfg;
- }
-
- /**
- * @throws Exception If failed.
- */
- public void testGroupingReducer() throws Exception {
- doTestGrouping(false);
- }
-
- /**
- * @throws Exception If failed.
- */
- public void testGroupingCombiner() throws Exception {
- doTestGrouping(true);
- }
-
- /**
- * @param combiner With combiner.
- * @throws Exception If failed.
- */
- public void doTestGrouping(boolean combiner) throws Exception {
- vals.clear();
-
- Job job = Job.getInstance();
-
- job.setInputFormatClass(InFormat.class);
- job.setOutputFormatClass(OutFormat.class);
-
- job.setOutputKeyClass(YearTemperature.class);
- job.setOutputValueClass(Text.class);
-
- job.setMapperClass(Mapper.class);
-
- if (combiner) {
- job.setCombinerClass(MyReducer.class);
- job.setNumReduceTasks(0);
- job.setCombinerKeyGroupingComparatorClass(YearComparator.class);
- }
- else {
- job.setReducerClass(MyReducer.class);
- job.setNumReduceTasks(4);
- job.setGroupingComparatorClass(YearComparator.class);
- }
-
- grid(0).hadoop().submit(new HadoopJobId(UUID.randomUUID(), 2),
- createJobInfo(job.getConfiguration())).get(30000);
-
- assertTrue(vals.isEmpty());
- }
-
- public static class MyReducer extends Reducer<YearTemperature, Text, Text, Object> {
- /** */
- int lastYear;
-
- @Override protected void reduce(YearTemperature key, Iterable<Text> vals0, Context context)
- throws IOException, InterruptedException {
- X.println("___ : " + context.getTaskAttemptID() + " --> " + key);
-
- Set<UUID> ids = new HashSet<>();
-
- for (Text val : vals0)
- assertTrue(ids.add(UUID.fromString(val.toString())));
-
- for (Text val : vals0)
- assertTrue(ids.remove(UUID.fromString(val.toString())));
-
- assertTrue(ids.isEmpty());
-
- assertTrue(key.year > lastYear);
-
- lastYear = key.year;
-
- for (Text val : vals0)
- assertTrue(vals.remove(UUID.fromString(val.toString())));
- }
- }
-
- public static class YearComparator implements RawComparator<YearTemperature> { // Grouping comparator.
- /** {@inheritDoc} */
- @Override public int compare(YearTemperature o1, YearTemperature o2) {
- return Integer.compare(o1.year, o2.year);
- }
-
- /** {@inheritDoc} */
- @Override public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {
- throw new IllegalStateException();
- }
- }
-
- public static class YearTemperature implements WritableComparable<YearTemperature>, Cloneable {
- /** */
- private int year;
-
- /** */
- private int temperature;
-
- /** {@inheritDoc} */
- @Override public void write(DataOutput out) throws IOException {
- out.writeInt(year);
- out.writeInt(temperature);
- }
-
- /** {@inheritDoc} */
- @Override public void readFields(DataInput in) throws IOException {
- year = in.readInt();
- temperature = in.readInt();
- }
-
- /** {@inheritDoc} */
- @Override public boolean equals(Object o) {
- throw new IllegalStateException();
- }
-
- /** {@inheritDoc} */
- @Override public int hashCode() { // To be partitioned by year.
- return year;
- }
-
- /** {@inheritDoc} */
- @Override public int compareTo(YearTemperature o) {
- int res = Integer.compare(year, o.year);
-
- if (res != 0)
- return res;
-
- // Sort comparator by year and temperature, to find max for year.
- return Integer.compare(o.temperature, temperature);
- }
-
- /** {@inheritDoc} */
- @Override public String toString() {
- return S.toString(YearTemperature.class, this);
- }
- }
-
- public static class InFormat extends InputFormat<YearTemperature, Text> {
- /** {@inheritDoc} */
- @Override public List<InputSplit> getSplits(JobContext context) throws IOException, InterruptedException {
- ArrayList<InputSplit> list = new ArrayList<>();
-
- for (int i = 0; i < 10; i++)
- list.add(new HadoopSortingTest.FakeSplit(20));
-
- return list;
- }
-
- /** {@inheritDoc} */
- @Override public RecordReader<YearTemperature, Text> createRecordReader(final InputSplit split,
- TaskAttemptContext context) throws IOException, InterruptedException {
- return new RecordReader<YearTemperature, Text>() {
- /** */
- int cnt;
-
- /** */
- Random rnd = new GridRandom();
-
- /** */
- YearTemperature key = new YearTemperature();
-
- /** */
- Text val = new Text();
-
- @Override public void initialize(InputSplit split, TaskAttemptContext context) {
- // No-op.
- }
-
- @Override public boolean nextKeyValue() throws IOException, InterruptedException {
- return cnt++ < split.getLength();
- }
-
- @Override public YearTemperature getCurrentKey() {
- key.year = 1990 + rnd.nextInt(10);
- key.temperature = 10 + rnd.nextInt(20);
-
- return key;
- }
-
- @Override public Text getCurrentValue() {
- UUID id = UUID.randomUUID();
-
- assertTrue(vals.add(id));
-
- val.set(id.toString());
-
- return val;
- }
-
- @Override public float getProgress() {
- return 0;
- }
-
- @Override public void close() {
- // No-op.
- }
- };
- }
- }
-
- /**
- *
- */
- public static class OutFormat extends OutputFormat {
- /** {@inheritDoc} */
- @Override public RecordWriter getRecordWriter(TaskAttemptContext context) throws IOException, InterruptedException {
- return null;
- }
-
- /** {@inheritDoc} */
- @Override public void checkOutputSpecs(JobContext context) throws IOException, InterruptedException {
- // No-op.
- }
-
- /** {@inheritDoc} */
- @Override public OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException, InterruptedException {
- return null;
- }
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/8032fc2c/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopJobTrackerSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopJobTrackerSelfTest.java b/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopJobTrackerSelfTest.java
deleted file mode 100644
index 9e268b7..0000000
--- a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopJobTrackerSelfTest.java
+++ /dev/null
@@ -1,345 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.processors.hadoop;
-
-import java.io.IOException;
-import java.net.URI;
-import java.net.URISyntaxException;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.UUID;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.atomic.AtomicInteger;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.mapreduce.InputFormat;
-import org.apache.hadoop.mapreduce.InputSplit;
-import org.apache.hadoop.mapreduce.Job;
-import org.apache.hadoop.mapreduce.JobContext;
-import org.apache.hadoop.mapreduce.Mapper;
-import org.apache.hadoop.mapreduce.RecordReader;
-import org.apache.hadoop.mapreduce.Reducer;
-import org.apache.hadoop.mapreduce.TaskAttemptContext;
-import org.apache.hadoop.mapreduce.lib.input.FileSplit;
-import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
-import org.apache.ignite.configuration.HadoopConfiguration;
-import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.IgniteKernal;
-import org.apache.ignite.internal.util.typedef.internal.U;
-
-import static org.apache.ignite.internal.processors.hadoop.HadoopUtils.createJobInfo;
-
-/**
- * Job tracker self test.
- */
-public class HadoopJobTrackerSelfTest extends HadoopAbstractSelfTest {
- /** */
- private static final String PATH_OUTPUT = "/test-out";
-
- /** Test block count parameter name. */
- private static final int BLOCK_CNT = 10;
-
- /** */
- private static HadoopSharedMap m = HadoopSharedMap.map(HadoopJobTrackerSelfTest.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 HadoopConfiguration hadoopConfiguration(String gridName) {
- HadoopConfiguration cfg = super.hadoopConfiguration(gridName);
-
- cfg.setMapReducePlanner(new HadoopTestRoundRobinMrPlanner());
-
- // TODO: IGNITE-404: Uncomment when fixed.
- //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"));
-
- HadoopJobId jobId = new HadoopJobId(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"));
-
- HadoopJobId jobId = new HadoopJobId(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(HadoopJobId jobId, boolean complete) throws Exception {
- for (int i = 0; i < gridCount(); i++) {
- IgniteKernal kernal = (IgniteKernal)grid(i);
-
- Hadoop hadoop = kernal.hadoop();
-
- HadoopJobStatus 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());
- }
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/8032fc2c/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceEmbeddedSelfTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceEmbeddedSelfTest.java b/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceEmbeddedSelfTest.java
deleted file mode 100644
index 25ef382..0000000
--- a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceEmbeddedSelfTest.java
+++ /dev/null
@@ -1,253 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.ignite.internal.processors.hadoop;
-
-import java.util.HashMap;
-import java.util.Map;
-import java.util.UUID;
-import org.apache.hadoop.conf.Configurable;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.CommonConfigurationKeys;
-import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.io.IntWritable;
-import org.apache.hadoop.io.Text;
-import org.apache.hadoop.io.serializer.WritableSerialization;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.JobConfigurable;
-import org.apache.hadoop.mapreduce.Job;
-import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
-import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
-import org.apache.ignite.configuration.HadoopConfiguration;
-import org.apache.ignite.igfs.IgfsPath;
-import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.processors.hadoop.examples.HadoopWordCount1;
-import org.apache.ignite.internal.processors.hadoop.examples.HadoopWordCount2;
-
-import static org.apache.ignite.internal.processors.hadoop.HadoopUtils.createJobInfo;
-
-/**
- * Tests map-reduce execution with embedded mode.
- */
-public class HadoopMapReduceEmbeddedSelfTest extends HadoopMapReduceTest {
- /** */
- private static Map<String, Boolean> flags = HadoopSharedMap.map(HadoopMapReduceEmbeddedSelfTest.class)
- .put("flags", new HashMap<String, Boolean>());
-
- /** {@inheritDoc} */
- @Override public HadoopConfiguration hadoopConfiguration(String gridName) {
- HadoopConfiguration cfg = super.hadoopConfiguration(gridName);
-
- // TODO: IGNITE-404: Uncomment when fixed.
- //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, HadoopWordCount2.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);
-
- HadoopWordCount1.setTasksClasses(jobConf, !useNewAPI, !useNewAPI, !useNewAPI);
-
- if (!useNewAPI) {
- jobConf.setPartitionerClass(CustomV1Partitioner.class);
- jobConf.setInputFormat(CustomV1InputFormat.class);
- jobConf.setOutputFormat(CustomV1OutputFormat.class);
- }
-
- Job job = Job.getInstance(jobConf);
-
- HadoopWordCount2.setTasksClasses(job, useNewAPI, useNewAPI, useNewAPI, false);
-
- 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(HadoopWordCount2.class);
-
- IgniteInternalFuture<?> fut = grid(0).hadoop().submit(new HadoopJobId(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);
- }
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/8032fc2c/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceErrorResilienceTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceErrorResilienceTest.java b/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceErrorResilienceTest.java
deleted file mode 100644
index dd12935..0000000
--- a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceErrorResilienceTest.java
+++ /dev/null
@@ -1,154 +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.ignite.igfs.IgfsPath;
-import org.apache.ignite.internal.processors.hadoop.examples.HadoopWordCount2;
-
-/**
- * Test of error resiliency after an error in a map-reduce job execution.
- * Combinations tested:
- * { new ALI, old API }
- * x { unchecked exception, checked exception, error }
- * x { phase where the error happens }.
- */
-public class HadoopMapReduceErrorResilienceTest extends HadoopAbstractMapReduceTest {
- /**
- * Tests recovery.
- *
- * @throws Exception If failed.
- */
- public void testRecoveryAfterAnError0_Runtime() throws Exception {
- doTestRecoveryAfterAnError(0, HadoopErrorSimulator.Kind.Runtime);
- }
-
- /**
- * Tests recovery.
- *
- * @throws Exception If failed.
- */
- public void testRecoveryAfterAnError0_IOException() throws Exception {
- doTestRecoveryAfterAnError(0, HadoopErrorSimulator.Kind.IOException);
- }
-
- /**
- * Tests recovery.
- *
- * @throws Exception If failed.
- */
- public void testRecoveryAfterAnError0_Error() throws Exception {
- doTestRecoveryAfterAnError(0, HadoopErrorSimulator.Kind.Error);
- }
-
- /**
- * Tests recovery.
- *
- * @throws Exception If failed.
- */
- public void testRecoveryAfterAnError7_Runtime() throws Exception {
- doTestRecoveryAfterAnError(7, HadoopErrorSimulator.Kind.Runtime);
- }
- /**
- * Tests recovery.
- *
- * @throws Exception If failed.
- */
- public void testRecoveryAfterAnError7_IOException() throws Exception {
- doTestRecoveryAfterAnError(7, HadoopErrorSimulator.Kind.IOException);
- }
- /**
- * Tests recovery.
- *
- * @throws Exception If failed.
- */
- public void testRecoveryAfterAnError7_Error() throws Exception {
- doTestRecoveryAfterAnError(7, HadoopErrorSimulator.Kind.Error);
- }
-
- /** {@inheritDoc} */
- @Override protected long getTestTimeout() {
- return 10 * 60 * 1000L;
- }
-
- /**
- * Tests correct work after an error.
- *
- * @throws Exception On error.
- */
- private void doTestRecoveryAfterAnError(int useNewBits, HadoopErrorSimulator.Kind simulatorKind) throws Exception {
- try {
- IgfsPath inDir = new IgfsPath(PATH_INPUT);
-
- igfs.mkdirs(inDir);
-
- IgfsPath inFile = new IgfsPath(inDir, HadoopWordCount2.class.getSimpleName() + "-input");
-
- generateTestFile(inFile.toString(), "red", red, "blue", blue, "green", green, "yellow", yellow);
-
- boolean useNewMapper = (useNewBits & 1) == 0;
- boolean useNewCombiner = (useNewBits & 2) == 0;
- boolean useNewReducer = (useNewBits & 4) == 0;
-
- for (int i = 0; i < 12; i++) {
- int bits = 1 << i;
-
- System.out.println("############################ Simulator kind = " + simulatorKind
- + ", Stage bits = " + bits);
-
- HadoopErrorSimulator sim = HadoopErrorSimulator.create(simulatorKind, bits);
-
- doTestWithErrorSimulator(sim, inFile, useNewMapper, useNewCombiner, useNewReducer);
- }
- } catch (Throwable t) {
- t.printStackTrace();
-
- fail("Unexpected throwable: " + t);
- }
- }
-
- /**
- * Performs test with given error simulator.
- *
- * @param sim The simulator.
- * @param inFile Input file.
- * @param useNewMapper If the use new mapper API.
- * @param useNewCombiner If to use new combiner.
- * @param useNewReducer If to use new reducer API.
- * @throws Exception If failed.
- */
- private void doTestWithErrorSimulator(HadoopErrorSimulator sim, IgfsPath inFile, boolean useNewMapper,
- boolean useNewCombiner, boolean useNewReducer) throws Exception {
- // Set real simulating error simulator:
- assertTrue(HadoopErrorSimulator.setInstance(HadoopErrorSimulator.noopInstance, sim));
-
- try {
- // Expect failure there:
- doTest(inFile, useNewMapper, useNewCombiner, useNewReducer);
- }
- catch (Throwable t) { // This may be an Error.
- // Expected:
- System.out.println(t.toString()); // Ignore, continue the test.
- }
-
- // Set no-op error simulator:
- assertTrue(HadoopErrorSimulator.setInstance(sim, HadoopErrorSimulator.noopInstance));
-
- // Expect success there:
- doTest(inFile, useNewMapper, useNewCombiner, useNewReducer);
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/8032fc2c/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceTest.java b/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceTest.java
deleted file mode 100644
index b703896..0000000
--- a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopMapReduceTest.java
+++ /dev/null
@@ -1,66 +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.ignite.igfs.IgfsPath;
-import org.apache.ignite.internal.processors.hadoop.examples.HadoopWordCount2;
-
-/**
- * Test of whole cycle of map-reduce processing via Job tracker.
- */
-public class HadoopMapReduceTest extends HadoopAbstractMapReduceTest {
- /**
- * 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, HadoopWordCount2.class.getSimpleName() + "-input");
-
- generateTestFile(inFile.toString(), "red", red, "blue", blue, "green", green, "yellow", yellow );
-
- for (boolean[] apiMode: getApiModes()) {
- assert apiMode.length == 3;
-
- boolean useNewMapper = apiMode[0];
- boolean useNewCombiner = apiMode[1];
- boolean useNewReducer = apiMode[2];
-
- doTest(inFile, useNewMapper, useNewCombiner, useNewReducer);
- }
- }
-
- /**
- * Gets API mode combinations to be tested.
- * Each boolean[] is { newMapper, newCombiner, newReducer } flag triplet.
- *
- * @return Arrays of booleans indicating API combinations to test.
- */
- protected boolean[][] getApiModes() {
- return new boolean[][] {
- { false, false, false },
- { false, false, true },
- { false, true, false },
- { true, false, false },
- { true, true, true },
- };
- }
-}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/8032fc2c/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopNoHadoopMapReduceTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopNoHadoopMapReduceTest.java b/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopNoHadoopMapReduceTest.java
deleted file mode 100644
index 0c172c3..0000000
--- a/modules/hadoop/src/test/java/org/apache/ignite/internal/processors/hadoop/HadoopNoHadoopMapReduceTest.java
+++ /dev/null
@@ -1,47 +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.ignite.configuration.IgniteConfiguration;
-
-/**
- * Test attempt to execute a map-reduce task while no Hadoop processor available.
- */
-public class HadoopNoHadoopMapReduceTest extends HadoopMapReduceTest {
- /** {@inheritDoc} */
- @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
- IgniteConfiguration c = super.getConfiguration(gridName);
-
- c.setHadoopConfiguration(null);
- c.setPeerClassLoadingEnabled(true);
-
- return c;
- }
-
- /** {@inheritDoc} */
- @Override public void testWholeMapReduceExecution() throws Exception {
- try {
- super.testWholeMapReduceExecution();
-
- fail("IllegalStateException expected.");
- }
- catch (IllegalStateException ignore) {
- // No-op.
- }
- }
-}