hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ste...@apache.org
Subject [03/14] hadoop git commit: HADOOP-13786 Add S3A committer for zero-rename commits to S3 endpoints. Contributed by Steve Loughran and Ryan Blue.
Date Wed, 22 Nov 2017 17:20:52 GMT
http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/TestMagicCommitPaths.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/TestMagicCommitPaths.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/TestMagicCommitPaths.java
new file mode 100644
index 0000000..47d112d
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/TestMagicCommitPaths.java
@@ -0,0 +1,246 @@
+/*
+ * 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.hadoop.fs.s3a.commit;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import com.google.common.collect.Lists;
+import org.junit.Assert;
+import org.junit.Test;
+
+import org.apache.hadoop.fs.Path;
+
+import static org.apache.hadoop.test.LambdaTestUtils.*;
+import static org.apache.hadoop.fs.s3a.commit.MagicCommitPaths.*;
+import static org.apache.hadoop.fs.s3a.commit.CommitConstants.*;
+
+/**
+ * Tests for {@link MagicCommitPaths} path operations.
+ */
+public class TestMagicCommitPaths extends Assert {
+
+  private static final List<String> MAGIC_AT_ROOT =
+      list(MAGIC);
+  private static final List<String> MAGIC_AT_ROOT_WITH_CHILD =
+      list(MAGIC, "child");
+  private static final List<String> MAGIC_WITH_CHILD =
+      list("parent", MAGIC, "child");
+  private static final List<String> MAGIC_AT_WITHOUT_CHILD =
+      list("parent", MAGIC);
+
+  private static final List<String> DEEP_MAGIC =
+      list("parent1", "parent2", MAGIC, "child1", "child2");
+
+  public static final String[] EMPTY = {};
+
+  @Test
+  public void testSplitPathEmpty() throws Throwable {
+    intercept(IllegalArgumentException.class,
+        () -> splitPathToElements(new Path("")));
+  }
+
+  @Test
+  public void testSplitPathDoubleBackslash() {
+    assertPathSplits("//", EMPTY);
+  }
+
+  @Test
+  public void testSplitRootPath() {
+    assertPathSplits("/", EMPTY);
+  }
+
+  @Test
+  public void testSplitBasic() {
+    assertPathSplits("/a/b/c",
+        new String[]{"a", "b", "c"});
+  }
+
+  @Test
+  public void testSplitTrailingSlash() {
+    assertPathSplits("/a/b/c/",
+        new String[]{"a", "b", "c"});
+  }
+
+  @Test
+  public void testSplitShortPath() {
+    assertPathSplits("/a",
+        new String[]{"a"});
+  }
+
+  @Test
+  public void testSplitShortPathTrailingSlash() {
+    assertPathSplits("/a/",
+        new String[]{"a"});
+  }
+
+  @Test
+  public void testParentsMagicRoot() {
+    assertParents(EMPTY, MAGIC_AT_ROOT);
+  }
+
+  @Test
+  public void testChildrenMagicRoot() {
+    assertChildren(EMPTY, MAGIC_AT_ROOT);
+  }
+
+  @Test
+  public void testParentsMagicRootWithChild() {
+    assertParents(EMPTY, MAGIC_AT_ROOT_WITH_CHILD);
+  }
+
+  @Test
+  public void testChildMagicRootWithChild() {
+    assertChildren(a("child"), MAGIC_AT_ROOT_WITH_CHILD);
+  }
+
+  @Test
+  public void testChildrenMagicWithoutChild() {
+    assertChildren(EMPTY, MAGIC_AT_WITHOUT_CHILD);
+  }
+
+  @Test
+  public void testChildMagicWithChild() {
+    assertChildren(a("child"), MAGIC_WITH_CHILD);
+  }
+
+  @Test
+  public void testParentMagicWithChild() {
+    assertParents(a("parent"), MAGIC_WITH_CHILD);
+  }
+
+  @Test
+  public void testParentDeepMagic() {
+    assertParents(a("parent1", "parent2"), DEEP_MAGIC);
+  }
+
+  @Test
+  public void testChildrenDeepMagic() {
+    assertChildren(a("child1", "child2"), DEEP_MAGIC);
+  }
+
+  @Test
+  public void testLastElementEmpty() throws Throwable {
+    intercept(IllegalArgumentException.class,
+        () -> lastElement(new ArrayList<>(0)));
+  }
+
+  @Test
+  public void testLastElementSingle() {
+    assertEquals("first", lastElement(l("first")));
+  }
+
+  @Test
+  public void testLastElementDouble() {
+    assertEquals("2", lastElement(l("first", "2")));
+  }
+
+  @Test
+  public void testFinalDestinationNoMagic() {
+    assertEquals(l("first", "2"),
+        finalDestination(l("first", "2")));
+  }
+
+  @Test
+  public void testFinalDestinationMagic1() {
+    assertEquals(l("first", "2"),
+        finalDestination(l("first", MAGIC, "2")));
+  }
+
+  @Test
+  public void testFinalDestinationMagic2() {
+    assertEquals(l("first", "3.txt"),
+        finalDestination(l("first", MAGIC, "2", "3.txt")));
+  }
+
+  @Test
+  public void testFinalDestinationRootMagic2() {
+    assertEquals(l("3.txt"),
+        finalDestination(l(MAGIC, "2", "3.txt")));
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testFinalDestinationMagicNoChild() {
+    finalDestination(l(MAGIC));
+  }
+
+  @Test
+  public void testFinalDestinationBaseDirectChild() {
+    finalDestination(l(MAGIC, BASE, "3.txt"));
+  }
+
+  @Test(expected = IllegalArgumentException.class)
+  public void testFinalDestinationBaseNoChild() {
+    assertEquals(l(), finalDestination(l(MAGIC, BASE)));
+  }
+
+  @Test
+  public void testFinalDestinationBaseSubdirsChild() {
+    assertEquals(l("2", "3.txt"),
+        finalDestination(l(MAGIC, "4", BASE, "2", "3.txt")));
+  }
+
+  /**
+   * If the base is above the magic dir, it's ignored.
+   */
+  @Test
+  public void testFinalDestinationIgnoresBaseBeforeMagic() {
+    assertEquals(l(BASE, "home", "3.txt"),
+        finalDestination(l(BASE, "home", MAGIC, "2", "3.txt")));
+  }
+
+  /** varargs to array. */
+  private static String[] a(String... str) {
+    return str;
+  }
+
+  /** list to array. */
+  private static List<String> l(String... str) {
+    return Arrays.asList(str);
+  }
+
+  /**
+   * Varags to list.
+   * @param args arguments
+   * @return a list
+   */
+  private static List<String> list(String... args) {
+    return Lists.newArrayList(args);
+  }
+
+  public void assertParents(String[] expected, List<String> elements) {
+    assertListEquals(expected, magicPathParents(elements));
+  }
+
+  public void assertChildren(String[] expected, List<String> elements) {
+    assertListEquals(expected, magicPathChildren(elements));
+  }
+
+  private void assertPathSplits(String pathString, String[] expected) {
+    Path path = new Path(pathString);
+    assertArrayEquals("From path " + path, expected,
+        splitPathToElements(path).toArray());
+  }
+
+  private void assertListEquals(String[] expected, List<String> actual) {
+    assertArrayEquals(expected, actual.toArray());
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/TestTasks.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/TestTasks.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/TestTasks.java
new file mode 100644
index 0000000..5e6fb82
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/TestTasks.java
@@ -0,0 +1,550 @@
+/*
+ * 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.hadoop.fs.s3a.commit;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.test.HadoopTestBase;
+
+import static org.apache.hadoop.test.LambdaTestUtils.intercept;
+
+/**
+ * Test Tasks class.
+ */
+@RunWith(Parameterized.class)
+public class TestTasks extends HadoopTestBase {
+  private static final Logger LOG = LoggerFactory.getLogger(TestTasks.class);
+  public static final int ITEM_COUNT = 16;
+  private static final int FAILPOINT = 8;
+
+  private final int numThreads;
+  /**
+   * Thread pool for task execution.
+   */
+  private ExecutorService threadPool;
+  private final CounterTask failingTask
+      = new CounterTask("failing committer", FAILPOINT, Item::commit);
+
+  private final FailureCounter failures
+      = new FailureCounter("failures", 0, null);
+  private final CounterTask reverter
+      = new CounterTask("reverter", 0, Item::revert);
+  private final CounterTask aborter
+      = new CounterTask("aborter", 0, Item::abort);
+
+  /**
+   * Test array for parameterized test runs: how many threads and
+   * to use. Threading makes some of the assertions brittle; there are
+   * more checks on single thread than parallel ops.
+   * @return a list of parameter tuples.
+   */
+  @Parameterized.Parameters
+  public static Collection<Object[]> params() {
+    return Arrays.asList(new Object[][]{
+        {0},
+        {1},
+        {3},
+        {8},
+        {16},
+    });
+  }
+
+
+  private List<Item> items;
+
+
+  /**
+   * Construct the parameterized test.
+   * @param numThreads number of threads
+   */
+  public TestTasks(int numThreads) {
+    this.numThreads = numThreads;
+  }
+
+  /**
+   * In a parallel test run there is more than one thread doing the execution.
+   * @return true if the threadpool size is >1
+   */
+  public boolean isParallel() {
+    return numThreads > 1;
+  }
+
+  @Before
+  public void setup() {
+    items = IntStream.rangeClosed(1, ITEM_COUNT)
+        .mapToObj(i -> new Item(i,
+            String.format("With %d threads", numThreads)))
+        .collect(Collectors.toList());
+
+    if (numThreads > 0) {
+      threadPool = Executors.newFixedThreadPool(numThreads,
+          new ThreadFactoryBuilder()
+              .setDaemon(true)
+              .setNameFormat(getMethodName() + "-pool-%d")
+              .build());
+    }
+
+  }
+
+  @After
+  public void teardown() {
+    if (threadPool != null) {
+      threadPool.shutdown();
+      threadPool = null;
+    }
+  }
+
+  /**
+   * create the builder.
+   * @return pre-inited builder
+   */
+  private Tasks.Builder<Item> builder() {
+    return Tasks.foreach(items).executeWith(threadPool);
+  }
+
+  private void assertRun(Tasks.Builder<Item> builder,
+      CounterTask task) throws IOException {
+    boolean b = builder.run(task);
+    assertTrue("Run of " + task + " failed", b);
+  }
+
+  private void assertFailed(Tasks.Builder<Item> builder,
+      CounterTask task) throws IOException {
+    boolean b = builder.run(task);
+    assertFalse("Run of " + task + " unexpectedly succeeded", b);
+  }
+
+  private String itemsToString() {
+    return "[" + items.stream().map(Item::toString)
+        .collect(Collectors.joining("\n")) +"]";
+  }
+
+  @Test
+  public void testSimpleInvocation() throws Throwable {
+    CounterTask t = new CounterTask("simple", 0, Item::commit);
+    assertRun(builder(), t);
+    t.assertInvoked("", ITEM_COUNT);
+  }
+
+  @Test
+  public void testFailNoStoppingSuppressed() throws Throwable {
+    assertFailed(builder().suppressExceptions(), failingTask);
+    failingTask.assertInvoked("Continued through operations", ITEM_COUNT);
+    items.forEach(Item::assertCommittedOrFailed);
+  }
+
+  @Test
+  public void testFailFastSuppressed() throws Throwable {
+    assertFailed(builder()
+            .suppressExceptions()
+            .stopOnFailure(),
+        failingTask);
+    if (isParallel()) {
+      failingTask.assertInvokedAtLeast("stop fast", FAILPOINT);
+    } else {
+      failingTask.assertInvoked("stop fast", FAILPOINT);
+    }
+  }
+
+  @Test
+  public void testFailedCallAbortSuppressed() throws Throwable {
+    assertFailed(builder()
+            .stopOnFailure()
+            .suppressExceptions()
+            .abortWith(aborter),
+        failingTask);
+    failingTask.assertInvokedAtLeast("success", FAILPOINT);
+    if (!isParallel()) {
+      aborter.assertInvokedAtLeast("abort", 1);
+      // all uncommitted items were aborted
+      items.stream().filter(i -> !i.committed)
+          .map(Item::assertAborted);
+      items.stream().filter(i -> i.committed)
+          .forEach(i -> assertFalse(i.toString(), i.aborted));
+    }
+  }
+
+  @Test
+  public void testFailedCalledWhenNotStoppingSuppressed() throws Throwable {
+    assertFailed(builder()
+            .suppressExceptions()
+            .onFailure(failures),
+        failingTask);
+    failingTask.assertInvokedAtLeast("success", FAILPOINT);
+    // only one failure was triggered
+    failures.assertInvoked("failure event", 1);
+  }
+
+  @Test
+  public void testFailFastCallRevertSuppressed() throws Throwable {
+    assertFailed(builder()
+            .stopOnFailure()
+            .revertWith(reverter)
+            .abortWith(aborter)
+            .suppressExceptions()
+            .onFailure(failures),
+        failingTask);
+    failingTask.assertInvokedAtLeast("success", FAILPOINT);
+    if (!isParallel()) {
+      aborter.assertInvokedAtLeast("abort", 1);
+      // all uncommitted items were aborted
+      items.stream().filter(i -> !i.committed)
+          .filter(i -> !i.failed)
+          .forEach(Item::assertAborted);
+    }
+    // all committed were reverted
+    items.stream().filter(i -> i.committed && !i.failed)
+        .forEach(Item::assertReverted);
+    // all reverted items are committed
+    items.stream().filter(i -> i.reverted)
+        .forEach(Item::assertCommitted);
+
+    // only one failure was triggered
+    failures.assertInvoked("failure event", 1);
+  }
+
+  @Test
+  public void testFailSlowCallRevertSuppressed() throws Throwable {
+    assertFailed(builder()
+            .suppressExceptions()
+            .revertWith(reverter)
+            .onFailure(failures),
+        failingTask);
+    failingTask.assertInvokedAtLeast("success", FAILPOINT);
+    // all committed were reverted
+    // identify which task failed from the set
+    int failing = failures.getItem().id;
+    items.stream()
+        .filter(i -> i.id != failing)
+        .filter(i -> i.committed)
+        .forEach(Item::assertReverted);
+    // all reverted items are committed
+    items.stream().filter(i -> i.reverted)
+        .forEach(Item::assertCommitted);
+
+    // only one failure was triggered
+    failures.assertInvoked("failure event", 1);
+  }
+
+  @Test
+  public void testFailFastExceptions() throws Throwable {
+    intercept(IOException.class,
+        () -> builder()
+            .stopOnFailure()
+            .run(failingTask));
+    if (isParallel()) {
+      failingTask.assertInvokedAtLeast("stop fast", FAILPOINT);
+    } else {
+      failingTask.assertInvoked("stop fast", FAILPOINT);
+    }
+  }
+
+  @Test
+  public void testFailSlowExceptions() throws Throwable {
+    intercept(IOException.class,
+        () -> builder()
+            .run(failingTask));
+    failingTask.assertInvoked("continued through operations", ITEM_COUNT);
+    items.forEach(Item::assertCommittedOrFailed);
+  }
+
+  @Test
+  public void testFailFastExceptionsWithAbortFailure() throws Throwable {
+    CounterTask failFirst = new CounterTask("task", 1, Item::commit);
+    CounterTask a = new CounterTask("aborter", 1, Item::abort);
+    intercept(IOException.class,
+        () -> builder()
+            .stopOnFailure()
+            .abortWith(a)
+            .run(failFirst));
+    if (!isParallel()) {
+      // expect the other tasks to be aborted
+      a.assertInvokedAtLeast("abort", ITEM_COUNT - 1);
+    }
+  }
+
+  @Test
+  public void testFailFastExceptionsWithAbortFailureStopped() throws Throwable {
+    CounterTask failFirst = new CounterTask("task", 1, Item::commit);
+    CounterTask a = new CounterTask("aborter", 1, Item::abort);
+    intercept(IOException.class,
+        () -> builder()
+            .stopOnFailure()
+            .stopAbortsOnFailure()
+            .abortWith(a)
+            .run(failFirst));
+    if (!isParallel()) {
+      // expect the other tasks to be aborted
+      a.assertInvoked("abort", 1);
+    }
+  }
+
+  /**
+   * Fail the last one committed, all the rest will be reverted.
+   * The actual ID of the last task has to be picke dup from the
+   * failure callback, as in the pool it may be one of any.
+   */
+  @Test
+  public void testRevertAllSuppressed() throws Throwable {
+    CounterTask failLast = new CounterTask("task", ITEM_COUNT, Item::commit);
+
+    assertFailed(builder()
+            .suppressExceptions()
+            .stopOnFailure()
+            .revertWith(reverter)
+            .abortWith(aborter)
+            .onFailure(failures),
+        failLast);
+    failLast.assertInvoked("success", ITEM_COUNT);
+    int abCount = aborter.getCount();
+    int revCount = reverter.getCount();
+    assertEquals(ITEM_COUNT, 1 + abCount + revCount);
+    // identify which task failed from the set
+    int failing = failures.getItem().id;
+    // all committed were reverted
+    items.stream()
+        .filter(i -> i.id != failing)
+        .filter(i -> i.committed)
+        .forEach(Item::assertReverted);
+    items.stream()
+        .filter(i -> i.id != failing)
+        .filter(i -> !i.committed)
+        .forEach(Item::assertAborted);
+    // all reverted items are committed
+    items.stream().filter(i -> i.reverted)
+        .forEach(Item::assertCommitted);
+
+    // only one failure was triggered
+    failures.assertInvoked("failure event", 1);
+  }
+
+
+  /**
+   * The Item which tasks process.
+   */
+  private final class Item {
+    private final int id;
+    private final String text;
+
+    private volatile boolean committed, aborted, reverted, failed;
+
+    private Item(int item, String text) {
+      this.id = item;
+      this.text = text;
+    }
+
+    boolean commit() {
+      committed = true;
+      return true;
+    }
+
+    boolean abort() {
+      aborted = true;
+      return true;
+    }
+
+    boolean revert() {
+      reverted = true;
+      return true;
+    }
+
+    boolean fail() {
+      failed = true;
+      return true;
+    }
+
+    public Item assertCommitted() {
+      assertTrue(toString() + " was not committed in\n"
+              +  itemsToString(),
+          committed);
+      return this;
+    }
+
+    public Item assertCommittedOrFailed() {
+      assertTrue(toString() + " was not committed nor failed in\n"
+              +  itemsToString(),
+          committed || failed);
+      return this;
+    }
+
+    public Item assertAborted() {
+      assertTrue(toString() + " was not aborted in\n"
+              + itemsToString(),
+          aborted);
+      return this;
+    }
+
+    public Item assertReverted() {
+      assertTrue(toString() + " was not reverted in\n"
+              + itemsToString(),
+          reverted);
+      return this;
+    }
+
+    @Override
+    public String toString() {
+      final StringBuilder sb = new StringBuilder("Item{");
+      sb.append(String.format("[%02d]", id));
+      sb.append(", committed=").append(committed);
+      sb.append(", aborted=").append(aborted);
+      sb.append(", reverted=").append(reverted);
+      sb.append(", failed=").append(failed);
+      sb.append(", text=").append(text);
+      sb.append('}');
+      return sb.toString();
+    }
+  }
+
+  /**
+   * Class which can count invocations and, if limit > 0, will raise
+   * an exception on the specific invocation of {@link #note(Object)}
+   * whose count == limit.
+   */
+  private class BaseCounter {
+    private final AtomicInteger counter = new AtomicInteger(0);
+    private final int limit;
+    private final String name;
+    private Item item;
+    private final Optional<Function<Item, Boolean>> action;
+
+    /**
+     * Base counter, tracks items.
+     * @param name name for string/exception/logs.
+     * @param limit limit at which an exception is raised, 0 == never
+     * @param action optional action to invoke after the increment,
+     * before limit check
+     */
+    BaseCounter(String name,
+        int limit,
+        Function<Item, Boolean> action) {
+      this.name = name;
+      this.limit = limit;
+      this.action = Optional.ofNullable(action);
+    }
+
+    /**
+     * Apply the action to an item; log at info afterwards with both the
+     * before and after string values of the item.
+     * @param i item to process.
+     * @throws IOException failure in the action
+     */
+    void process(Item i) throws IOException {
+      this.item = i;
+      int count = counter.incrementAndGet();
+      if (limit == count) {
+        i.fail();
+        LOG.info("{}: Failed {}", this, i);
+        throw new IOException(String.format("%s: Limit %d reached for %s",
+            this, limit, i));
+      }
+      String before = i.toString();
+      action.map(a -> a.apply(i));
+      LOG.info("{}: {} -> {}", this, before, i);
+    }
+
+    int getCount() {
+      return counter.get();
+    }
+
+    Item getItem() {
+      return item;
+    }
+
+    void assertInvoked(String text, int expected) {
+      assertEquals(toString() + ": " + text, expected, getCount());
+    }
+
+    void assertInvokedAtLeast(String text, int expected) {
+      int actual = getCount();
+      assertTrue(toString() + ": " + text
+              + "-expected " + expected
+              + " invocations, but got " + actual
+              + " in " + itemsToString(),
+          expected <= actual);
+    }
+
+    @Override
+    public String toString() {
+      final StringBuilder sb = new StringBuilder(
+          "BaseCounter{");
+      sb.append("name='").append(name).append('\'');
+      sb.append(", count=").append(counter.get());
+      sb.append(", limit=").append(limit);
+      sb.append(", item=").append(item);
+      sb.append('}');
+      return sb.toString();
+    }
+  }
+
+  private final class CounterTask
+      extends BaseCounter implements Tasks.Task<Item, IOException> {
+
+    private CounterTask(String name, int limit,
+        Function<Item, Boolean> action) {
+      super(name, limit, action);
+    }
+
+    @Override
+    public void run(Item item) throws IOException {
+      process(item);
+    }
+
+  }
+
+  private final class FailureCounter
+      extends BaseCounter implements Tasks.FailureTask<Item, IOException> {
+    private Exception exception;
+
+    private FailureCounter(String name, int limit,
+        Function<Item, Boolean> action) {
+      super(name, limit, action);
+    }
+
+    @Override
+    public void run(Item item, Exception ex) throws IOException {
+      process(item);
+      this.exception = ex;
+    }
+
+    private Exception getException() {
+      return exception;
+    }
+
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/magic/ITMagicCommitMRJob.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/magic/ITMagicCommitMRJob.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/magic/ITMagicCommitMRJob.java
new file mode 100644
index 0000000..57eb8b2
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/magic/ITMagicCommitMRJob.java
@@ -0,0 +1,70 @@
+/*
+ * 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.hadoop.fs.s3a.commit.magic;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.s3a.commit.AbstractITCommitMRJob;
+import org.apache.hadoop.fs.s3a.commit.files.SuccessData;
+
+import static org.apache.hadoop.fs.s3a.commit.CommitConstants.*;
+
+/**
+ * Full integration test for the Magic Committer.
+ *
+ * There's no need to disable the committer setting for the filesystem here,
+ * because the committers are being instantiated in their own processes;
+ * the settings in {@link #applyCustomConfigOptions(Configuration)} are
+ * passed down to these processes.
+ */
+public class ITMagicCommitMRJob extends AbstractITCommitMRJob {
+
+  /**
+   * Need consistency here.
+   * @return false
+   */
+  @Override
+  public boolean useInconsistentClient() {
+    return false;
+  }
+
+  @Override
+  protected String committerName() {
+    return MagicS3GuardCommitter.NAME;
+  }
+
+  /**
+   * Turn on the magic commit support for the FS, else nothing will work.
+   * @param conf configuration
+   */
+  @Override
+  protected void applyCustomConfigOptions(Configuration conf) {
+    conf.setBoolean(MAGIC_COMMITTER_ENABLED, true);
+  }
+
+  /**
+   * Check that the magic dir was cleaned up.
+   * {@inheritDoc}
+   */
+  @Override
+  protected void customPostExecutionValidation(Path destPath,
+      SuccessData successData) throws Exception {
+    assertPathDoesNotExist("No cleanup", new Path(destPath, MAGIC));
+  }
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/magic/ITestMagicCommitProtocol.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/magic/ITestMagicCommitProtocol.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/magic/ITestMagicCommitProtocol.java
new file mode 100644
index 0000000..74c1d9d
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/magic/ITestMagicCommitProtocol.java
@@ -0,0 +1,190 @@
+/*
+ * 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.hadoop.fs.s3a.commit.magic;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.contract.ContractTestUtils;
+import org.apache.hadoop.fs.s3a.commit.AbstractITCommitProtocol;
+import org.apache.hadoop.fs.s3a.commit.AbstractS3ACommitter;
+import org.apache.hadoop.fs.s3a.commit.CommitConstants;
+import org.apache.hadoop.fs.s3a.commit.CommitUtils;
+import org.apache.hadoop.fs.s3a.commit.CommitterFaultInjection;
+import org.apache.hadoop.fs.s3a.commit.CommitterFaultInjectionImpl;
+import org.apache.hadoop.mapreduce.JobContext;
+import org.apache.hadoop.mapreduce.JobStatus;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+
+import java.io.IOException;
+
+import static org.apache.hadoop.fs.s3a.commit.CommitConstants.*;
+
+/**
+ * Test the magic committer's commit protocol.
+ */
+public class ITestMagicCommitProtocol extends AbstractITCommitProtocol {
+
+  @Override
+  protected String suitename() {
+    return "ITestMagicCommitProtocol";
+  }
+
+  /**
+   * Need consistency here.
+   * @return false
+   */
+  @Override
+  public boolean useInconsistentClient() {
+    return false;
+  }
+
+  @Override
+  protected Configuration createConfiguration() {
+    Configuration conf = super.createConfiguration();
+    conf.setBoolean(MAGIC_COMMITTER_ENABLED, true);
+    return conf;
+  }
+
+  @Override
+  protected String getCommitterFactoryName() {
+    return CommitConstants.S3A_COMMITTER_FACTORY;
+  }
+
+  @Override
+  protected String getCommitterName() {
+    return CommitConstants.COMMITTER_NAME_MAGIC;
+  }
+
+  @Override
+  public void setup() throws Exception {
+    super.setup();
+    CommitUtils.verifyIsMagicCommitFS(getFileSystem());
+  }
+
+  @Override
+  public void assertJobAbortCleanedUp(JobData jobData)
+      throws Exception {
+    // special handling of magic directory; harmless in staging
+    Path magicDir = new Path(getOutDir(), MAGIC);
+    ContractTestUtils.assertPathDoesNotExist(getFileSystem(),
+        "magic dir ", magicDir);
+    super.assertJobAbortCleanedUp(jobData);
+  }
+
+  @Override
+  protected AbstractS3ACommitter createCommitter(
+      Path outputPath,
+      TaskAttemptContext context)
+      throws IOException {
+    return new MagicS3GuardCommitter(outputPath, context);
+  }
+
+  public AbstractS3ACommitter createFailingCommitter(
+      TaskAttemptContext tContext) throws IOException {
+    return new CommitterWithFailedThenSucceed(getOutDir(), tContext);
+  }
+
+  protected void validateTaskAttemptPathDuringWrite(Path p) throws IOException {
+    String pathStr = p.toString();
+    assertTrue("not magic " + pathStr,
+        pathStr.contains(MAGIC));
+    assertPathDoesNotExist("task attempt visible", p);
+  }
+
+  protected void validateTaskAttemptPathAfterWrite(Path p) throws IOException {
+    FileStatus st = getFileSystem().getFileStatus(p);
+    assertEquals("file length in " + st, 0, st.getLen());
+    Path pendingFile = new Path(p.toString() + PENDING_SUFFIX);
+    assertPathExists("pending file", pendingFile);
+  }
+
+  /**
+   * The class provides a overridden implementation of commitJobInternal which
+   * causes the commit failed for the first time then succeed.
+   */
+
+  private static final class CommitterWithFailedThenSucceed extends
+      MagicS3GuardCommitter implements CommitterFaultInjection {
+    private final CommitterFaultInjectionImpl injection;
+
+    CommitterWithFailedThenSucceed(Path outputPath,
+        TaskAttemptContext context) throws IOException {
+      super(outputPath, context);
+      injection = new CommitterFaultInjectionImpl(outputPath, context, true);
+    }
+
+    @Override
+    public void setupJob(JobContext context) throws IOException {
+      injection.setupJob(context);
+      super.setupJob(context);
+    }
+
+    @Override
+    public void abortJob(JobContext context, JobStatus.State state)
+        throws IOException {
+      injection.abortJob(context, state);
+      super.abortJob(context, state);
+    }
+
+    @Override
+    @SuppressWarnings("deprecation")
+    public void cleanupJob(JobContext context) throws IOException {
+      injection.cleanupJob(context);
+      super.cleanupJob(context);
+    }
+
+    @Override
+    public void setupTask(TaskAttemptContext context) throws IOException {
+      injection.setupTask(context);
+      super.setupTask(context);
+    }
+
+    @Override
+    public void commitTask(TaskAttemptContext context) throws IOException {
+      injection.commitTask(context);
+      super.commitTask(context);
+    }
+
+    @Override
+    public void abortTask(TaskAttemptContext context) throws IOException {
+      injection.abortTask(context);
+      super.abortTask(context);
+    }
+
+    @Override
+    public void commitJob(JobContext context) throws IOException {
+      injection.commitJob(context);
+      super.commitJob(context);
+    }
+
+    @Override
+    public boolean needsTaskCommit(TaskAttemptContext context)
+        throws IOException {
+      injection.needsTaskCommit(context);
+      return super.needsTaskCommit(context);
+    }
+
+    @Override
+    public void setFaults(CommitterFaultInjection.Faults... faults) {
+      injection.setFaults(faults);
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/magic/ITestS3AHugeMagicCommits.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/magic/ITestS3AHugeMagicCommits.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/magic/ITestS3AHugeMagicCommits.java
new file mode 100644
index 0000000..e3a295b
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/magic/ITestS3AHugeMagicCommits.java
@@ -0,0 +1,195 @@
+/*
+ * 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.hadoop.fs.s3a.commit.magic;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.LocatedFileStatus;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.contract.ContractTestUtils;
+import org.apache.hadoop.fs.s3a.Constants;
+import org.apache.hadoop.fs.s3a.S3AFileSystem;
+import org.apache.hadoop.fs.s3a.commit.CommitConstants;
+import org.apache.hadoop.fs.s3a.commit.CommitOperations;
+import org.apache.hadoop.fs.s3a.commit.CommitUtils;
+import org.apache.hadoop.fs.s3a.commit.files.PendingSet;
+import org.apache.hadoop.fs.s3a.commit.files.SinglePendingCommit;
+import org.apache.hadoop.fs.s3a.scale.AbstractSTestS3AHugeFiles;
+
+import static org.apache.hadoop.fs.s3a.commit.CommitConstants.*;
+import static org.apache.hadoop.fs.s3a.S3ATestUtils.*;
+
+
+/**
+ * Write a huge file via the magic commit mechanism,
+ * commit it and verify that it is there. This is needed to
+ * verify that the pending-upload mechanism works with multipart files
+ * of more than one part.
+ *
+ * This is a scale test.
+ */
+public class ITestS3AHugeMagicCommits extends AbstractSTestS3AHugeFiles {
+  private static final Logger LOG = LoggerFactory.getLogger(
+      ITestS3AHugeMagicCommits.class);
+
+  private Path magicDir;
+  private Path jobDir;
+
+  /** file used as the destination for the write;
+   *  it is never actually created. */
+  private Path magicOutputFile;
+
+  /** The file with the JSON data about the commit. */
+  private Path pendingDataFile;
+
+  /**
+   * Use fast upload on disk.
+   * @return the upload buffer mechanism.
+   */
+  protected String getBlockOutputBufferName() {
+    return Constants.FAST_UPLOAD_BUFFER_DISK;
+  }
+
+  /**
+   * The suite name; required to be unique.
+   * @return the test suite name
+   */
+  @Override
+  public String getTestSuiteName() {
+    return "ITestS3AHugeMagicCommits";
+  }
+
+  /**
+   * Create the scale IO conf with the committer enabled.
+   * @return the configuration to use for the test FS.
+   */
+  @Override
+  protected Configuration createScaleConfiguration() {
+    Configuration conf = super.createScaleConfiguration();
+    conf.setBoolean(MAGIC_COMMITTER_ENABLED, true);
+    return conf;
+  }
+
+  @Override
+  public void setup() throws Exception {
+    super.setup();
+    CommitUtils.verifyIsMagicCommitFS(getFileSystem());
+
+    // set up the paths for the commit operation
+    Path finalDirectory = new Path(getScaleTestDir(), "commit");
+    magicDir = new Path(finalDirectory, MAGIC);
+    jobDir = new Path(magicDir, "job_001");
+    String filename = "commit.bin";
+    setHugefile(new Path(finalDirectory, filename));
+    magicOutputFile = new Path(jobDir, filename);
+    pendingDataFile = new Path(jobDir, filename + PENDING_SUFFIX);
+  }
+
+  /**
+   * Returns the path to the commit metadata file, not that of the huge file.
+   * @return a file in the job dir
+   */
+  @Override
+  protected Path getPathOfFileToCreate() {
+    return magicOutputFile;
+  }
+
+  @Override
+  public void test_030_postCreationAssertions() throws Throwable {
+    describe("Committing file");
+    assertPathDoesNotExist("final file exists", getHugefile());
+    assertPathExists("No pending file", pendingDataFile);
+    S3AFileSystem fs = getFileSystem();
+
+    // as a 0-byte marker is created, there is a file at the end path,
+    // it just MUST be 0-bytes long
+    FileStatus status = fs.getFileStatus(magicOutputFile);
+    assertEquals("Non empty marker file " + status,
+        0, status.getLen());
+    ContractTestUtils.NanoTimer timer = new ContractTestUtils.NanoTimer();
+    CommitOperations operations = new CommitOperations(fs);
+    Path destDir = getHugefile().getParent();
+    assertPathExists("Magic dir", new Path(destDir, CommitConstants.MAGIC));
+    String destDirKey = fs.pathToKey(destDir);
+    List<String> uploads = listMultipartUploads(fs, destDirKey);
+
+    assertEquals("Pending uploads: "
+        + uploads.stream()
+        .collect(Collectors.joining("\n")), 1, uploads.size());
+    assertNotNull("jobDir", jobDir);
+    Pair<PendingSet, List<Pair<LocatedFileStatus, IOException>>>
+        results = operations.loadSinglePendingCommits(jobDir, false);
+    for (SinglePendingCommit singlePendingCommit :
+        results.getKey().getCommits()) {
+      operations.commitOrFail(singlePendingCommit);
+    }
+    timer.end("time to commit %s", pendingDataFile);
+    // upload is no longer pending
+    uploads = listMultipartUploads(fs, destDirKey);
+    assertEquals("Pending uploads"
+            + uploads.stream().collect(Collectors.joining("\n")),
+        0, operations.listPendingUploadsUnderPath(destDir).size());
+    // at this point, the huge file exists, so the normal assertions
+    // on that file must be valid. Verify.
+    super.test_030_postCreationAssertions();
+  }
+
+  private void skipQuietly(String text) {
+    describe("Skipping: %s", text);
+  }
+
+  @Override
+  public void test_040_PositionedReadHugeFile() {
+    skipQuietly("test_040_PositionedReadHugeFile");
+  }
+
+  @Override
+  public void test_050_readHugeFile() {
+    skipQuietly("readHugeFile");
+  }
+
+  @Override
+  public void test_100_renameHugeFile() {
+    skipQuietly("renameHugeFile");
+  }
+
+  @Override
+  public void test_800_DeleteHugeFiles() throws IOException {
+    if (getFileSystem() != null) {
+      try {
+        getFileSystem().abortOutstandingMultipartUploads(0);
+      } catch (IOException e) {
+        LOG.info("Exception while purging old uploads", e);
+      }
+    }
+    try {
+      super.test_800_DeleteHugeFiles();
+    } finally {
+      ContractTestUtils.rm(getFileSystem(), magicDir, true, false);
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/staging/MockedStagingCommitter.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/staging/MockedStagingCommitter.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/staging/MockedStagingCommitter.java
new file mode 100644
index 0000000..47383b7
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/staging/MockedStagingCommitter.java
@@ -0,0 +1,98 @@
+/*
+ * 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.hadoop.fs.s3a.commit.staging;
+
+import java.io.IOException;
+import java.io.ObjectOutputStream;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.s3a.MockS3AFileSystem;
+import org.apache.hadoop.fs.s3a.commit.staging.StagingTestBase.ClientErrors;
+import org.apache.hadoop.fs.s3a.commit.staging.StagingTestBase.ClientResults;
+import org.apache.hadoop.mapreduce.JobContext;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+
+/**
+ * Committer subclass that uses a mocked S3A connection for testing.
+ */
+class MockedStagingCommitter extends StagingCommitter {
+
+  MockedStagingCommitter(Path outputPath,
+      TaskAttemptContext context)
+      throws IOException {
+    super(outputPath, context);
+  }
+
+  /**
+   * Returns the mock FS without checking FS type.
+   * @param out output path
+   * @param config job/task config
+   * @return a filesystem.
+   * @throws IOException IO failure
+   */
+  @Override
+  protected FileSystem getDestinationFS(Path out, Configuration config)
+      throws IOException {
+    return out.getFileSystem(config);
+  }
+
+  @Override
+  public void commitJob(JobContext context) throws IOException {
+    // turn off stamping an output marker, as that codepath isn't mocked yet.
+    super.commitJob(context);
+    Configuration conf = context.getConfiguration();
+    try {
+      String jobCommitterPath = conf.get("mock-results-file");
+      if (jobCommitterPath != null) {
+        try (ObjectOutputStream out = new ObjectOutputStream(
+            FileSystem.getLocal(conf)
+                .create(new Path(jobCommitterPath), false))) {
+          out.writeObject(getResults());
+        }
+      }
+    } catch (Exception e) {
+      // do nothing, the test will fail
+    }
+  }
+
+  @Override
+  protected void maybeCreateSuccessMarker(JobContext context,
+      List<String> filenames)
+      throws IOException {
+     //skipped
+  }
+
+  public ClientResults getResults() throws IOException {
+    MockS3AFileSystem mockFS = (MockS3AFileSystem)getDestS3AFS();
+    return mockFS.getOutcome().getKey();
+  }
+
+  public ClientErrors getErrors() throws IOException {
+    MockS3AFileSystem mockFS = (MockS3AFileSystem) getDestS3AFS();
+    return mockFS.getOutcome().getValue();
+  }
+
+  @Override
+  public String toString() {
+    return "MockedStagingCommitter{ " + super.toString() + " ";
+  }
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/staging/PartitionedCommitterForTesting.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/staging/PartitionedCommitterForTesting.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/staging/PartitionedCommitterForTesting.java
new file mode 100644
index 0000000..5adce15
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/staging/PartitionedCommitterForTesting.java
@@ -0,0 +1,58 @@
+/*
+ * 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.hadoop.fs.s3a.commit.staging;
+
+import java.io.IOException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+
+/**
+ * Partitioned committer overridden for better testing.
+ */
+class PartitionedCommitterForTesting extends
+    PartitionedStagingCommitter {
+
+  PartitionedCommitterForTesting(Path outputPath,
+      TaskAttemptContext context) throws IOException {
+    super(outputPath, context);
+  }
+
+  @Override
+  protected void initOutput(Path out) throws IOException {
+    super.initOutput(out);
+    setOutputPath(out);
+  }
+
+  /**
+   * Returns the mock FS without checking FS type.
+   * @param out output path
+   * @param config job/task config
+   * @return a filesystem.
+   * @throws IOException failure to get the FS
+   */
+  @Override
+  protected FileSystem getDestinationFS(Path out, Configuration config)
+      throws IOException {
+    return out.getFileSystem(config);
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/staging/StagingTestBase.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/staging/StagingTestBase.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/staging/StagingTestBase.java
new file mode 100644
index 0000000..38d5156
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/staging/StagingTestBase.java
@@ -0,0 +1,724 @@
+/*
+ * 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.hadoop.fs.s3a.commit.staging;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.io.Serializable;
+import java.net.URI;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+import com.amazonaws.AmazonClientException;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.AmazonS3Client;
+import com.amazonaws.services.s3.model.AbortMultipartUploadRequest;
+import com.amazonaws.services.s3.model.AmazonS3Exception;
+import com.amazonaws.services.s3.model.CompleteMultipartUploadRequest;
+import com.amazonaws.services.s3.model.CompleteMultipartUploadResult;
+import com.amazonaws.services.s3.model.DeleteObjectRequest;
+import com.amazonaws.services.s3.model.InitiateMultipartUploadRequest;
+import com.amazonaws.services.s3.model.InitiateMultipartUploadResult;
+import com.amazonaws.services.s3.model.ListMultipartUploadsRequest;
+import com.amazonaws.services.s3.model.MultipartUpload;
+import com.amazonaws.services.s3.model.MultipartUploadListing;
+import com.amazonaws.services.s3.model.UploadPartRequest;
+import com.amazonaws.services.s3.model.UploadPartResult;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FileSystemTestHelper;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.s3a.Constants;
+import org.apache.hadoop.fs.s3a.MockS3AFileSystem;
+import org.apache.hadoop.fs.s3a.S3AFileSystem;
+import org.apache.hadoop.fs.s3a.commit.AbstractCommitITest;
+import org.apache.hadoop.fs.s3a.commit.CommitConstants;
+import org.apache.hadoop.fs.s3a.commit.InternalCommitterConstants;
+import org.apache.hadoop.fs.s3a.commit.MiniDFSClusterService;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapreduce.JobContext;
+import org.apache.hadoop.mapreduce.JobID;
+import org.apache.hadoop.mapreduce.OutputCommitter;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.TaskAttemptID;
+import org.apache.hadoop.mapreduce.TaskID;
+import org.apache.hadoop.mapreduce.TaskType;
+import org.apache.hadoop.mapreduce.task.JobContextImpl;
+import org.apache.hadoop.mapreduce.task.TaskAttemptContextImpl;
+import org.apache.hadoop.mapreduce.v2.util.MRBuilderUtils;
+import org.apache.hadoop.service.ServiceOperations;
+import org.apache.hadoop.test.HadoopTestBase;
+
+import static org.mockito.Matchers.*;
+import static org.mockito.Mockito.*;
+
+/**
+ * Test base for mock tests of staging committers:
+ * core constants and static methods, inner classes
+ * for specific test types.
+ *
+ * Some of the verification methods here are unused...they are being left
+ * in place in case changes on the implementation make the verifications
+ * relevant again.
+ */
+public class StagingTestBase {
+  private static final Logger LOG =
+      LoggerFactory.getLogger(StagingTestBase.class);
+
+  public static final String BUCKET = MockS3AFileSystem.BUCKET;
+  public static final String OUTPUT_PREFIX = "output/path";
+  public static final Path OUTPUT_PATH =
+      new Path("s3a://" + BUCKET + "/" + OUTPUT_PREFIX);
+  public static final URI OUTPUT_PATH_URI = OUTPUT_PATH.toUri();
+
+  protected StagingTestBase() {
+  }
+
+  /**
+   * Sets up the mock filesystem instance and binds it to the
+   * {@link FileSystem#get(URI, Configuration)} call for the supplied URI
+   * and config.
+   * All standard mocking setup MUST go here.
+   * @param conf config to use
+   * @param outcome tuple of outcomes to store in mock FS
+   * @return the filesystem created
+   * @throws IOException IO problems.
+   */
+  protected static S3AFileSystem createAndBindMockFSInstance(Configuration conf,
+      Pair<StagingTestBase.ClientResults, StagingTestBase.ClientErrors> outcome)
+      throws IOException {
+    S3AFileSystem mockFs = mockS3AFileSystemRobustly();
+    MockS3AFileSystem wrapperFS = new MockS3AFileSystem(mockFs, outcome);
+    URI uri = OUTPUT_PATH_URI;
+    wrapperFS.initialize(uri, conf);
+    FileSystemTestHelper.addFileSystemForTesting(uri, conf, wrapperFS);
+    return mockFs;
+  }
+
+  private static S3AFileSystem mockS3AFileSystemRobustly() {
+    S3AFileSystem mockFS = mock(S3AFileSystem.class);
+    doNothing().when(mockFS).incrementReadOperations();
+    doNothing().when(mockFS).incrementWriteOperations();
+    doNothing().when(mockFS).incrementWriteOperations();
+    doNothing().when(mockFS).incrementWriteOperations();
+    return mockFS;
+  }
+
+  /**
+   * Look up the FS by URI, return a (cast) Mock wrapper.
+   * @param conf config
+   * @return the FS
+   * @throws IOException IO Failure
+   */
+  public static MockS3AFileSystem lookupWrapperFS(Configuration conf)
+      throws IOException {
+    return (MockS3AFileSystem) FileSystem.get(OUTPUT_PATH_URI, conf);
+  }
+
+  public static void verifyCompletion(FileSystem mockS3) throws IOException {
+    verifyCleanupTempFiles(mockS3);
+    verifyNoMoreInteractions(mockS3);
+  }
+
+  public static void verifyDeleted(FileSystem mockS3, Path path)
+      throws IOException {
+    verify(mockS3).delete(path, true);
+  }
+
+  public static void verifyDeleted(FileSystem mockS3, String child)
+      throws IOException {
+    verifyDeleted(mockS3, new Path(OUTPUT_PATH, child));
+  }
+
+  public static void verifyCleanupTempFiles(FileSystem mockS3)
+      throws IOException {
+    verifyDeleted(mockS3,
+        new Path(OUTPUT_PATH, CommitConstants.TEMPORARY));
+  }
+
+  protected static void assertConflictResolution(
+      StagingCommitter committer,
+      JobContext job,
+      ConflictResolution mode) {
+    Assert.assertEquals("Conflict resolution mode in " + committer,
+        mode, committer.getConflictResolutionMode(job, new Configuration()));
+  }
+
+  public static void pathsExist(FileSystem mockS3, String... children)
+      throws IOException {
+    for (String child : children) {
+      pathExists(mockS3, new Path(OUTPUT_PATH, child));
+    }
+  }
+
+  public static void pathExists(FileSystem mockS3, Path path)
+      throws IOException {
+    when(mockS3.exists(path)).thenReturn(true);
+  }
+
+  public static void pathDoesNotExist(FileSystem mockS3, Path path)
+      throws IOException {
+    when(mockS3.exists(path)).thenReturn(false);
+  }
+
+  public static void canDelete(FileSystem mockS3, String... children)
+      throws IOException {
+    for (String child : children) {
+      canDelete(mockS3, new Path(OUTPUT_PATH, child));
+    }
+  }
+
+  public static void canDelete(FileSystem mockS3, Path f) throws IOException {
+    when(mockS3.delete(f,
+        true /* recursive */))
+        .thenReturn(true);
+  }
+
+  public static void verifyExistenceChecked(FileSystem mockS3, String child)
+      throws IOException {
+    verifyExistenceChecked(mockS3, new Path(OUTPUT_PATH, child));
+  }
+
+  public static void verifyExistenceChecked(FileSystem mockS3, Path path)
+      throws IOException {
+    verify(mockS3).exists(path);
+  }
+
+  /**
+   * Provides setup/teardown of a MiniDFSCluster for tests that need one.
+   */
+  public static class MiniDFSTest extends HadoopTestBase {
+
+    private static MiniDFSClusterService hdfs;
+
+    private static JobConf conf = null;
+
+    protected static JobConf getConfiguration() {
+      return conf;
+    }
+
+    protected static FileSystem getDFS() {
+      return hdfs.getClusterFS();
+    }
+
+    /**
+     * Setup the mini HDFS cluster.
+     * @throws IOException Failure
+     */
+    @BeforeClass
+    @SuppressWarnings("deprecation")
+    public static void setupHDFS() throws IOException {
+      if (hdfs == null) {
+        JobConf c = new JobConf();
+        hdfs = new MiniDFSClusterService();
+        hdfs.init(c);
+        hdfs.start();
+        conf = c;
+      }
+    }
+
+    @SuppressWarnings("ThrowableNotThrown")
+    @AfterClass
+    public static void teardownFS() throws IOException {
+      ServiceOperations.stopQuietly(hdfs);
+      conf = null;
+      hdfs = null;
+    }
+
+  }
+
+  /**
+   * Base class for job committer tests.
+   * @param <C> committer
+   */
+  public abstract static class JobCommitterTest<C extends OutputCommitter>
+      extends HadoopTestBase {
+    private static final JobID JOB_ID = new JobID("job", 1);
+    private JobConf jobConf;
+
+    // created in BeforeClass
+    private S3AFileSystem mockFS = null;
+    private MockS3AFileSystem wrapperFS = null;
+    private JobContext job = null;
+
+    // created in Before
+    private StagingTestBase.ClientResults results = null;
+    private StagingTestBase.ClientErrors errors = null;
+    private AmazonS3 mockClient = null;
+
+    @Before
+    public void setupJob() throws Exception {
+      this.jobConf = new JobConf();
+      jobConf.set(InternalCommitterConstants.FS_S3A_COMMITTER_STAGING_UUID,
+          UUID.randomUUID().toString());
+      jobConf.setBoolean(
+          CommitConstants.CREATE_SUCCESSFUL_JOB_OUTPUT_DIR_MARKER,
+          false);
+
+      this.job = new JobContextImpl(jobConf, JOB_ID);
+      this.results = new StagingTestBase.ClientResults();
+      this.errors = new StagingTestBase.ClientErrors();
+      this.mockClient = newMockS3Client(results, errors);
+      this.mockFS = createAndBindMockFSInstance(jobConf,
+          Pair.of(results, errors));
+      this.wrapperFS = lookupWrapperFS(jobConf);
+      // and bind the FS
+      wrapperFS.setAmazonS3Client(mockClient);
+    }
+
+    public S3AFileSystem getMockS3A() {
+      return mockFS;
+    }
+
+    public MockS3AFileSystem getWrapperFS() {
+      return wrapperFS;
+    }
+
+    public JobContext getJob() {
+      return job;
+    }
+
+    /**
+     * Create a task attempt for a job by creating a stub task ID.
+     * @return a task attempt
+     */
+    public TaskAttemptContext createTaskAttemptForJob() {
+      return AbstractCommitITest.taskAttemptForJob(
+          MRBuilderUtils.newJobId(1, JOB_ID.getId(), 1), job);
+    }
+
+    protected StagingTestBase.ClientResults getMockResults() {
+      return results;
+    }
+
+    protected StagingTestBase.ClientErrors getMockErrors() {
+      return errors;
+    }
+
+    abstract C newJobCommitter() throws Exception;
+  }
+
+  /** Abstract test of task commits. */
+  public abstract static class TaskCommitterTest<C extends OutputCommitter>
+      extends JobCommitterTest<C> {
+    private static final TaskAttemptID AID = new TaskAttemptID(
+        new TaskID(JobCommitterTest.JOB_ID, TaskType.REDUCE, 2), 3);
+
+    private C jobCommitter = null;
+    private TaskAttemptContext tac = null;
+    private File tempDir;
+
+    @Before
+    public void setupTask() throws Exception {
+      this.jobCommitter = newJobCommitter();
+      jobCommitter.setupJob(getJob());
+
+      this.tac = new TaskAttemptContextImpl(
+          new Configuration(getJob().getConfiguration()), AID);
+
+      // get the task's configuration copy so modifications take effect
+      String tmp = System.getProperty(
+          StagingCommitterConstants.JAVA_IO_TMPDIR);
+      tempDir = new File(tmp);
+      tac.getConfiguration().set(Constants.BUFFER_DIR, tmp + "/buffer");
+      tac.getConfiguration().set(
+          CommitConstants.FS_S3A_COMMITTER_STAGING_TMP_PATH,
+          tmp + "/cluster");
+    }
+
+    protected C getJobCommitter() {
+      return jobCommitter;
+    }
+
+    protected TaskAttemptContext getTAC() {
+      return tac;
+    }
+
+    abstract C newTaskCommitter() throws Exception;
+
+    protected File getTempDir() {
+      return tempDir;
+    }
+  }
+
+  /**
+   * Results accrued during mock runs.
+   * This data is serialized in MR Tests and read back in in the test runner
+   */
+  public static class ClientResults implements Serializable {
+    private static final long serialVersionUID = -3137637327090709905L;
+    // For inspection of what the committer did
+    private final Map<String, InitiateMultipartUploadRequest> requests =
+        Maps.newHashMap();
+    private final List<String> uploads = Lists.newArrayList();
+    private final List<UploadPartRequest> parts = Lists.newArrayList();
+    private final Map<String, List<String>> tagsByUpload = Maps.newHashMap();
+    private final List<CompleteMultipartUploadRequest> commits =
+        Lists.newArrayList();
+    private final List<AbortMultipartUploadRequest> aborts
+        = Lists.newArrayList();
+    private final Map<String, String> activeUploads =
+        Maps.newHashMap();
+    private final List<DeleteObjectRequest> deletes = Lists.newArrayList();
+
+    public Map<String, InitiateMultipartUploadRequest> getRequests() {
+      return requests;
+    }
+
+    public List<String> getUploads() {
+      return uploads;
+    }
+
+    public List<UploadPartRequest> getParts() {
+      return parts;
+    }
+
+    public Map<String, List<String>> getTagsByUpload() {
+      return tagsByUpload;
+    }
+
+    public List<CompleteMultipartUploadRequest> getCommits() {
+      return commits;
+    }
+
+    public List<AbortMultipartUploadRequest> getAborts() {
+      return aborts;
+    }
+
+    public List<DeleteObjectRequest> getDeletes() {
+      return deletes;
+    }
+
+    public void resetDeletes() {
+      deletes.clear();
+    }
+
+    public void resetUploads() {
+      uploads.clear();
+      activeUploads.clear();
+    }
+
+    public void resetCommits() {
+      commits.clear();
+    }
+
+    public void resetRequests() {
+      requests.clear();
+    }
+
+    @Override
+    public String toString() {
+      final StringBuilder sb = new StringBuilder(
+          super.toString());
+      sb.append("{ requests=").append(requests.size());
+      sb.append(", uploads=").append(uploads.size());
+      sb.append(", parts=").append(parts.size());
+      sb.append(", tagsByUpload=").append(tagsByUpload.size());
+      sb.append(", commits=").append(commits.size());
+      sb.append(", aborts=").append(aborts.size());
+      sb.append(", deletes=").append(deletes.size());
+      sb.append('}');
+      return sb.toString();
+    }
+  }
+
+  /** Control errors to raise in mock S3 client. */
+  public static class ClientErrors {
+    // For injecting errors
+    private int failOnInit = -1;
+    private int failOnUpload = -1;
+    private int failOnCommit = -1;
+    private int failOnAbort = -1;
+    private boolean recover = false;
+
+    public void failOnInit(int initNum) {
+      this.failOnInit = initNum;
+    }
+
+    public void failOnUpload(int uploadNum) {
+      this.failOnUpload = uploadNum;
+    }
+
+    public void failOnCommit(int commitNum) {
+      this.failOnCommit = commitNum;
+    }
+
+    public void failOnAbort(int abortNum) {
+      this.failOnAbort = abortNum;
+    }
+
+    public void recoverAfterFailure() {
+      this.recover = true;
+    }
+
+    @Override
+    public String toString() {
+      final StringBuilder sb = new StringBuilder(
+          "ClientErrors{");
+      sb.append("failOnInit=").append(failOnInit);
+      sb.append(", failOnUpload=").append(failOnUpload);
+      sb.append(", failOnCommit=").append(failOnCommit);
+      sb.append(", failOnAbort=").append(failOnAbort);
+      sb.append(", recover=").append(recover);
+      sb.append('}');
+      return sb.toString();
+    }
+
+    public int getFailOnInit() {
+      return failOnInit;
+    }
+
+    public int getFailOnUpload() {
+      return failOnUpload;
+    }
+
+    public int getFailOnCommit() {
+      return failOnCommit;
+    }
+
+    public int getFailOnAbort() {
+      return failOnAbort;
+    }
+
+    public boolean isRecover() {
+      return recover;
+    }
+  }
+
+  /**
+   * Instantiate mock client with the results and errors requested.
+   * @param results results to accrue
+   * @param errors when (if any) to fail
+   * @return the mock client to patch in to a committer/FS instance
+   */
+  public static AmazonS3 newMockS3Client(final ClientResults results,
+      final ClientErrors errors) {
+    AmazonS3Client mockClient = mock(AmazonS3Client.class);
+    final Object lock = new Object();
+
+    // initiateMultipartUpload
+    when(mockClient
+        .initiateMultipartUpload(any(InitiateMultipartUploadRequest.class)))
+        .thenAnswer(invocation -> {
+          LOG.debug("initiateMultipartUpload for {}", mockClient);
+          synchronized (lock) {
+            if (results.requests.size() == errors.failOnInit) {
+              if (errors.recover) {
+                errors.failOnInit(-1);
+              }
+              throw new AmazonClientException(
+                  "Mock Fail on init " + results.requests.size());
+            }
+            String uploadId = UUID.randomUUID().toString();
+            InitiateMultipartUploadRequest req = invocation.getArgumentAt(
+                0, InitiateMultipartUploadRequest.class);
+            results.requests.put(uploadId, req);
+            results.activeUploads.put(uploadId, req.getKey());
+            results.uploads.add(uploadId);
+            return newResult(results.requests.get(uploadId), uploadId);
+          }
+        });
+
+    // uploadPart
+    when(mockClient.uploadPart(any(UploadPartRequest.class)))
+        .thenAnswer(invocation -> {
+          LOG.debug("uploadPart for {}", mockClient);
+          synchronized (lock) {
+            if (results.parts.size() == errors.failOnUpload) {
+              if (errors.recover) {
+                errors.failOnUpload(-1);
+              }
+              LOG.info("Triggering upload failure");
+              throw new AmazonClientException(
+                  "Mock Fail on upload " + results.parts.size());
+            }
+            UploadPartRequest req = invocation.getArgumentAt(
+                0, UploadPartRequest.class);
+            results.parts.add(req);
+            String etag = UUID.randomUUID().toString();
+            List<String> etags = results.tagsByUpload.get(req.getUploadId());
+            if (etags == null) {
+              etags = Lists.newArrayList();
+              results.tagsByUpload.put(req.getUploadId(), etags);
+            }
+            etags.add(etag);
+            return newResult(req, etag);
+          }
+        });
+
+    // completeMultipartUpload
+    when(mockClient
+        .completeMultipartUpload(any(CompleteMultipartUploadRequest.class)))
+        .thenAnswer(invocation -> {
+          LOG.debug("completeMultipartUpload for {}", mockClient);
+          synchronized (lock) {
+            if (results.commits.size() == errors.failOnCommit) {
+              if (errors.recover) {
+                errors.failOnCommit(-1);
+              }
+              throw new AmazonClientException(
+                  "Mock Fail on commit " + results.commits.size());
+            }
+            CompleteMultipartUploadRequest req = invocation.getArgumentAt(
+                0, CompleteMultipartUploadRequest.class);
+            results.commits.add(req);
+            results.activeUploads.remove(req.getUploadId());
+
+            return newResult(req);
+          }
+        });
+
+    // abortMultipartUpload mocking
+    doAnswer(invocation -> {
+      LOG.debug("abortMultipartUpload for {}", mockClient);
+      synchronized (lock) {
+        if (results.aborts.size() == errors.failOnAbort) {
+          if (errors.recover) {
+            errors.failOnAbort(-1);
+          }
+          throw new AmazonClientException(
+              "Mock Fail on abort " + results.aborts.size());
+        }
+        AbortMultipartUploadRequest req = invocation.getArgumentAt(
+            0, AbortMultipartUploadRequest.class);
+        String id = req.getUploadId();
+        String p = results.activeUploads.remove(id);
+        if (p == null) {
+          // upload doesn't exist
+          AmazonS3Exception ex = new AmazonS3Exception(
+              "not found " + id);
+          ex.setStatusCode(404);
+          throw ex;
+        }
+        results.aborts.add(req);
+        return null;
+      }
+    })
+        .when(mockClient)
+        .abortMultipartUpload(any(AbortMultipartUploadRequest.class));
+
+    // deleteObject mocking
+    doAnswer(invocation -> {
+      LOG.debug("deleteObject for {}", mockClient);
+      synchronized (lock) {
+        results.deletes.add(invocation.getArgumentAt(
+            0, DeleteObjectRequest.class));
+        return null;
+      }
+    })
+        .when(mockClient)
+        .deleteObject(any(DeleteObjectRequest.class));
+
+    // deleteObject mocking
+    doAnswer(invocation -> {
+      LOG.debug("deleteObject for {}", mockClient);
+      synchronized (lock) {
+        results.deletes.add(new DeleteObjectRequest(
+            invocation.getArgumentAt(0, String.class),
+            invocation.getArgumentAt(1, String.class)
+        ));
+        return null;
+      }
+    }).when(mockClient)
+        .deleteObject(any(String.class), any(String.class));
+
+    // to String returns the debug information
+    when(mockClient.toString()).thenAnswer(
+        invocation -> "Mock3AClient " + results + " " + errors);
+
+    when(mockClient
+        .listMultipartUploads(any(ListMultipartUploadsRequest.class)))
+        .thenAnswer(invocation -> {
+          synchronized (lock) {
+            MultipartUploadListing l = new MultipartUploadListing();
+            l.setMultipartUploads(
+                results.activeUploads.entrySet().stream()
+                    .map(e -> newMPU(e.getKey(), e.getValue()))
+                    .collect(Collectors.toList()));
+            return l;
+          }
+        });
+
+    return mockClient;
+  }
+
+  private static CompleteMultipartUploadResult newResult(
+      CompleteMultipartUploadRequest req) {
+    return new CompleteMultipartUploadResult();
+  }
+
+
+  private static MultipartUpload newMPU(String id, String path) {
+    MultipartUpload up = new MultipartUpload();
+    up.setUploadId(id);
+    up.setKey(path);
+    return up;
+  }
+
+  private static UploadPartResult newResult(UploadPartRequest request,
+      String etag) {
+    UploadPartResult result = new UploadPartResult();
+    result.setPartNumber(request.getPartNumber());
+    result.setETag(etag);
+    return result;
+  }
+
+  private static InitiateMultipartUploadResult newResult(
+      InitiateMultipartUploadRequest request, String uploadId) {
+    InitiateMultipartUploadResult result = new InitiateMultipartUploadResult();
+    result.setUploadId(uploadId);
+    return result;
+  }
+
+  /**
+   * create files in the attempt path that should be found by
+   * {@code getTaskOutput}.
+   * @param relativeFiles list of files relative to address path
+   * @param attemptPath attempt path
+   * @param conf config for FS
+   * @throws IOException on any failure
+   */
+  public static void createTestOutputFiles(List<String> relativeFiles,
+      Path attemptPath,
+      Configuration conf) throws IOException {
+    //
+    FileSystem attemptFS = attemptPath.getFileSystem(conf);
+    attemptFS.delete(attemptPath, true);
+    for (String relative : relativeFiles) {
+      // 0-length files are ignored, so write at least one byte
+      OutputStream out = attemptFS.create(new Path(attemptPath, relative));
+      out.write(34);
+      out.close();
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/staging/TestPaths.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/staging/TestPaths.java b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/staging/TestPaths.java
new file mode 100644
index 0000000..ee6480a
--- /dev/null
+++ b/hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/staging/TestPaths.java
@@ -0,0 +1,127 @@
+/*
+ * 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.hadoop.fs.s3a.commit.staging;
+
+import org.junit.Test;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.LocalFileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.test.HadoopTestBase;
+
+import static org.apache.hadoop.fs.s3a.commit.staging.Paths.*;
+import static org.apache.hadoop.test.LambdaTestUtils.intercept;
+
+/**
+ * Test {@link org.apache.hadoop.fs.s3a.commit.staging.Paths}.
+ */
+public class TestPaths extends HadoopTestBase {
+
+  @Test
+  public void testUUIDPart() {
+    assertUUIDAdded("/part-0000", "/part-0000-UUID");
+  }
+
+  @Test
+  public void testUUIDPartSuffix() {
+    assertUUIDAdded("/part-0000.gz.csv", "/part-0000-UUID.gz.csv");
+  }
+
+  @Test
+  public void testUUIDDottedPath() {
+    assertUUIDAdded("/parent.dir/part-0000", "/parent.dir/part-0000-UUID");
+  }
+
+  @Test
+  public void testUUIDPartUUID() {
+    assertUUIDAdded("/part-0000-UUID.gz.csv", "/part-0000-UUID.gz.csv");
+  }
+
+  @Test
+  public void testUUIDParentUUID() {
+    assertUUIDAdded("/UUID/part-0000.gz.csv", "/UUID/part-0000.gz.csv");
+  }
+
+  @Test
+  public void testUUIDDir() throws Throwable {
+    intercept(IllegalStateException.class,
+        () -> addUUID("/dest/", "UUID"));
+  }
+
+  @Test
+  public void testUUIDEmptyDir() throws Throwable {
+    intercept(IllegalArgumentException.class,
+        () -> addUUID("", "UUID"));
+  }
+
+  @Test
+  public void testEmptyUUID() throws Throwable {
+    intercept(IllegalArgumentException.class,
+        () -> addUUID("part-0000.gz", ""));
+  }
+
+  private void assertUUIDAdded(String path, String expected) {
+    assertEquals("from " + path, expected, addUUID(path, "UUID"));
+  }
+
+  private static final String DATA = "s3a://landsat-pds/data/";
+  private static final Path BASE = new Path(DATA);
+
+  @Test
+  public void testRelativizeOneLevel() {
+    String suffix = "year=2017";
+    Path path = new Path(DATA + suffix);
+    assertEquals(suffix, getRelativePath(BASE, path));
+  }
+
+  @Test
+  public void testRelativizeTwoLevel() {
+    String suffix = "year=2017/month=10";
+    Path path = path(BASE, suffix);
+    assertEquals(suffix, getRelativePath(BASE, path));
+  }
+
+  @Test
+  public void testRelativizeSelf() {
+    assertEquals("", getRelativePath(BASE, BASE));
+  }
+
+  @Test
+  public void testRelativizeParent() {
+    // goes up to the parent if one is above the other
+    assertEquals("/", getRelativePath(BASE, BASE.getParent()));
+  }
+
+  @Test
+  public void testGetPartition() {
+    assertEquals("year=2017/month=10",
+        getPartition("year=2017/month=10/part-0000.avro"));
+  }
+
+  @Test
+  public void testMPUCommitDir() throws Throwable {
+    Configuration conf = new Configuration();
+    LocalFileSystem localFS = FileSystem.getLocal(conf);
+    Path dir = getMultipartUploadCommitsDirectory(localFS, conf, "UUID");
+    assertTrue(dir.toString().endsWith("UUID/"
+        + StagingCommitterConstants.STAGING_UPLOADS));
+  }
+
+}


---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org


Mime
View raw message