You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by kk...@apache.org on 2017/11/28 21:47:45 UTC
[27/50] [abbrv] hadoop git commit: HADOOP-13786 Add S3A committer for
zero-rename commits to S3 endpoints. Contributed by Steve Loughran and Ryan
Blue.
HADOOP-13786 Add S3A committer for zero-rename commits to S3 endpoints.
Contributed by Steve Loughran and Ryan Blue.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/de8b6ca5
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/de8b6ca5
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/de8b6ca5
Branch: refs/heads/YARN-6592
Commit: de8b6ca5ef8614de6d6277b7617e27c788b0555c
Parents: 782ba3b
Author: Steve Loughran <st...@apache.org>
Authored: Wed Nov 22 15:28:12 2017 +0000
Committer: Steve Loughran <st...@apache.org>
Committed: Wed Nov 22 15:28:12 2017 +0000
----------------------------------------------------------------------
.../dev-support/findbugsExcludeFile.xml | 7 +
.../apache/hadoop/fs/FSDataOutputStream.java | 9 +
.../apache/hadoop/fs/PathExistsException.java | 4 +-
.../org/apache/hadoop/fs/StorageStatistics.java | 5 +
.../apache/hadoop/util/JsonSerialization.java | 299 +++
.../src/main/resources/core-default.xml | 117 +-
.../hadoop/fs/contract/ContractTestUtils.java | 51 +-
.../apache/hadoop/test/GenericTestUtils.java | 29 +-
.../org/apache/hadoop/test/HadoopTestBase.java | 51 +-
.../org/apache/hadoop/test/LambdaTestUtils.java | 144 +-
.../hadoop/util/TestJsonSerialization.java | 185 ++
.../mapreduce/TestMapreduceConfigFields.java | 27 +-
.../lib/output/BindingPathOutputCommitter.java | 184 ++
.../lib/output/FileOutputCommitter.java | 12 +-
.../lib/output/FileOutputCommitterFactory.java | 38 +
.../mapreduce/lib/output/FileOutputFormat.java | 10 +-
.../lib/output/NamedCommitterFactory.java | 79 +
.../lib/output/PathOutputCommitter.java | 17 +
.../lib/output/PathOutputCommitterFactory.java | 204 ++
.../src/main/resources/mapred-default.xml | 22 +
.../lib/output/TestPathOutputCommitter.java | 24 +-
.../output/TestPathOutputCommitterFactory.java | 495 +++++
hadoop-tools/hadoop-aws/pom.xml | 46 +-
.../hadoop/fs/s3a/AWSBadRequestException.java | 42 +
.../hadoop/fs/s3a/AWSClientIOException.java | 3 +-
.../hadoop/fs/s3a/AWSNoResponseException.java | 31 +
.../hadoop/fs/s3a/AWSRedirectException.java | 38 +
.../fs/s3a/AWSServiceThrottledException.java | 42 +
.../hadoop/fs/s3a/AWSStatus500Exception.java | 37 +
.../s3a/BlockingThreadPoolExecutorService.java | 2 +-
.../org/apache/hadoop/fs/s3a/Constants.java | 72 +-
.../fs/s3a/InconsistentAmazonS3Client.java | 232 ++-
.../java/org/apache/hadoop/fs/s3a/Invoker.java | 485 +++++
.../java/org/apache/hadoop/fs/s3a/Listing.java | 26 +-
.../java/org/apache/hadoop/fs/s3a/Retries.java | 92 +
.../hadoop/fs/s3a/S3ABlockOutputStream.java | 307 +--
.../org/apache/hadoop/fs/s3a/S3ADataBlocks.java | 2 +-
.../org/apache/hadoop/fs/s3a/S3AFileSystem.java | 940 +++++----
.../apache/hadoop/fs/s3a/S3AInputStream.java | 56 +-
.../hadoop/fs/s3a/S3AInstrumentation.java | 231 ++-
.../apache/hadoop/fs/s3a/S3ARetryPolicy.java | 246 +++
.../hadoop/fs/s3a/S3AStorageStatistics.java | 12 +-
.../java/org/apache/hadoop/fs/s3a/S3AUtils.java | 324 ++-
.../org/apache/hadoop/fs/s3a/S3ListRequest.java | 11 +
.../hadoop/fs/s3a/S3ObjectAttributes.java | 10 +-
.../org/apache/hadoop/fs/s3a/Statistic.java | 56 +-
.../hadoop/fs/s3a/WriteOperationHelper.java | 474 +++++
.../fs/s3a/commit/AbstractS3ACommitter.java | 756 +++++++
.../s3a/commit/AbstractS3ACommitterFactory.java | 90 +
.../hadoop/fs/s3a/commit/CommitConstants.java | 240 +++
.../hadoop/fs/s3a/commit/CommitOperations.java | 596 ++++++
.../hadoop/fs/s3a/commit/CommitUtils.java | 129 ++
.../hadoop/fs/s3a/commit/CommitUtilsWithMR.java | 192 ++
.../apache/hadoop/fs/s3a/commit/Duration.java | 60 +
.../hadoop/fs/s3a/commit/DurationInfo.java | 59 +
.../s3a/commit/InternalCommitterConstants.java | 100 +
.../hadoop/fs/s3a/commit/LocalTempDir.java | 80 +
.../fs/s3a/commit/MagicCommitIntegration.java | 182 ++
.../hadoop/fs/s3a/commit/MagicCommitPaths.java | 229 ++
.../fs/s3a/commit/PathCommitException.java | 43 +
.../apache/hadoop/fs/s3a/commit/PutTracker.java | 100 +
.../fs/s3a/commit/S3ACommitterFactory.java | 129 ++
.../org/apache/hadoop/fs/s3a/commit/Tasks.java | 410 ++++
.../hadoop/fs/s3a/commit/ValidationFailure.java | 53 +
.../hadoop/fs/s3a/commit/files/PendingSet.java | 192 ++
.../s3a/commit/files/PersistentCommitData.java | 69 +
.../s3a/commit/files/SinglePendingCommit.java | 432 ++++
.../hadoop/fs/s3a/commit/files/SuccessData.java | 322 +++
.../fs/s3a/commit/files/package-info.java | 45 +
.../fs/s3a/commit/magic/MagicCommitTracker.java | 161 ++
.../s3a/commit/magic/MagicS3GuardCommitter.java | 288 +++
.../magic/MagicS3GuardCommitterFactory.java | 47 +
.../fs/s3a/commit/magic/package-info.java | 27 +
.../hadoop/fs/s3a/commit/package-info.java | 28 +
.../s3a/commit/staging/ConflictResolution.java | 33 +
.../staging/DirectoryStagingCommitter.java | 116 ++
.../DirectoryStagingCommitterFactory.java | 48 +
.../staging/PartitionedStagingCommitter.java | 159 ++
.../PartitionedStagingCommitterFactory.java | 48 +
.../hadoop/fs/s3a/commit/staging/Paths.java | 300 +++
.../fs/s3a/commit/staging/StagingCommitter.java | 851 ++++++++
.../staging/StagingCommitterConstants.java | 64 +
.../commit/staging/StagingCommitterFactory.java | 49 +
.../fs/s3a/commit/staging/package-info.java | 27 +
.../fs/s3a/s3guard/DynamoDBMetadataStore.java | 368 ++--
.../apache/hadoop/fs/s3a/s3guard/S3Guard.java | 2 +
.../hadoop/fs/s3a/s3guard/S3GuardTool.java | 16 +-
.../tools/hadoop-aws/committer_architecture.md | 1951 ++++++++++++++++++
.../markdown/tools/hadoop-aws/committers.md | 819 ++++++++
.../src/site/markdown/tools/hadoop-aws/index.md | 188 +-
.../site/markdown/tools/hadoop-aws/s3guard.md | 7 +-
.../site/markdown/tools/hadoop-aws/testing.md | 60 +
.../tools/hadoop-aws/troubleshooting_s3a.md | 124 ++
.../s3a/ITestS3AContractGetFileStatus.java | 11 +
.../hadoop/fs/s3a/AbstractS3AMockTest.java | 5 +-
.../hadoop/fs/s3a/AbstractS3ATestBase.java | 23 +-
.../hadoop/fs/s3a/ITestS3AConfiguration.java | 8 +-
.../hadoop/fs/s3a/ITestS3AEncryptionSSEC.java | 6 +-
.../hadoop/fs/s3a/ITestS3AFailureHandling.java | 49 +-
.../fs/s3a/ITestS3AFileOperationCost.java | 2 +
.../fs/s3a/ITestS3ATemporaryCredentials.java | 15 +-
.../fs/s3a/ITestS3GuardListConsistency.java | 45 +-
.../apache/hadoop/fs/s3a/MockS3AFileSystem.java | 322 +++
.../hadoop/fs/s3a/MockS3ClientFactory.java | 8 +
.../apache/hadoop/fs/s3a/S3ATestConstants.java | 6 +
.../org/apache/hadoop/fs/s3a/S3ATestUtils.java | 132 +-
.../hadoop/fs/s3a/StorageStatisticsTracker.java | 99 +
.../org/apache/hadoop/fs/s3a/TestInvoker.java | 460 +++++
.../org/apache/hadoop/fs/s3a/TestListing.java | 2 +-
.../fs/s3a/TestS3AExceptionTranslation.java | 84 +-
.../fs/s3a/commit/AbstractCommitITest.java | 412 ++++
.../fs/s3a/commit/AbstractITCommitMRJob.java | 324 +++
.../fs/s3a/commit/AbstractITCommitProtocol.java | 1371 ++++++++++++
.../fs/s3a/commit/CommitterFaultInjection.java | 44 +
.../s3a/commit/CommitterFaultInjectionImpl.java | 131 ++
.../fs/s3a/commit/ITestCommitOperations.java | 545 +++++
.../fs/s3a/commit/LoggingTextOutputFormat.java | 133 ++
.../fs/s3a/commit/MiniDFSClusterService.java | 79 +
.../fs/s3a/commit/TestMagicCommitPaths.java | 246 +++
.../apache/hadoop/fs/s3a/commit/TestTasks.java | 550 +++++
.../fs/s3a/commit/magic/ITMagicCommitMRJob.java | 70 +
.../commit/magic/ITestMagicCommitProtocol.java | 190 ++
.../commit/magic/ITestS3AHugeMagicCommits.java | 195 ++
.../commit/staging/MockedStagingCommitter.java | 98 +
.../staging/PartitionedCommitterForTesting.java | 58 +
.../fs/s3a/commit/staging/StagingTestBase.java | 724 +++++++
.../hadoop/fs/s3a/commit/staging/TestPaths.java | 127 ++
.../commit/staging/TestStagingCommitter.java | 696 +++++++
.../TestStagingDirectoryOutputCommitter.java | 138 ++
.../TestStagingPartitionedFileListing.java | 186 ++
.../TestStagingPartitionedJobCommit.java | 236 +++
.../TestStagingPartitionedTaskCommit.java | 237 +++
.../integration/ITDirectoryCommitMRJob.java | 33 +
.../integration/ITPartitionCommitMRJob.java | 33 +
.../integration/ITStagingCommitMRJob.java | 66 +
.../ITestDirectoryCommitProtocol.java | 131 ++
.../ITestPartitionedCommitProtocol.java | 139 ++
.../integration/ITestStagingCommitProtocol.java | 190 ++
.../s3guard/AbstractS3GuardToolTestBase.java | 23 +
.../s3a/s3guard/TestDynamoDBMetadataStore.java | 13 +-
.../fs/s3a/scale/AbstractSTestS3AHugeFiles.java | 191 +-
.../hadoop-aws/src/test/resources/core-site.xml | 9 +-
.../src/test/resources/log4j.properties | 42 +-
.../registry/client/binding/JsonSerDeser.java | 224 +-
144 files changed, 24108 insertions(+), 1172 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-common-project/hadoop-common/dev-support/findbugsExcludeFile.xml
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/dev-support/findbugsExcludeFile.xml b/hadoop-common-project/hadoop-common/dev-support/findbugsExcludeFile.xml
index 4bafd8e..c056d21 100644
--- a/hadoop-common-project/hadoop-common/dev-support/findbugsExcludeFile.xml
+++ b/hadoop-common-project/hadoop-common/dev-support/findbugsExcludeFile.xml
@@ -446,4 +446,11 @@
<Method name="setInstance"/>
<Bug pattern="ME_ENUM_FIELD_SETTER"/>
</Match>
+
+ <!-- findbugs is complaining that a stream isn't closed. It will be. -->
+ <Match>
+ <Class name="org.apache.hadoop.util.JsonSerialization"/>
+ <Method name="save"/>
+ <Bug pattern="OBL_UNSATISFIED_OBLIGATION"/>
+ </Match>
</FindBugsFilter>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSDataOutputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSDataOutputStream.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSDataOutputStream.java
index 1d95cd3..5970373 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSDataOutputStream.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSDataOutputStream.java
@@ -101,6 +101,15 @@ public class FSDataOutputStream extends DataOutputStream
out.close(); // This invokes PositionCache.close()
}
+ @Override
+ public String toString() {
+ final StringBuilder sb = new StringBuilder(
+ "FSDataOutputStream{");
+ sb.append("wrappedStream=").append(wrappedStream);
+ sb.append('}');
+ return sb.toString();
+ }
+
/**
* Get a reference to the wrapped output stream.
*
http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/PathExistsException.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/PathExistsException.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/PathExistsException.java
index ccc1f0c..cd9f70a 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/PathExistsException.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/PathExistsException.java
@@ -27,7 +27,7 @@ public class PathExistsException extends PathIOException {
super(path, "File exists");
}
- protected PathExistsException(String path, String error) {
+ public PathExistsException(String path, String error) {
super(path, error);
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/StorageStatistics.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/StorageStatistics.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/StorageStatistics.java
index d987ad0..5a3d736 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/StorageStatistics.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/StorageStatistics.java
@@ -97,6 +97,11 @@ public abstract class StorageStatistics {
public long getValue() {
return value;
}
+
+ @Override
+ public String toString() {
+ return name + " = " + value;
+ }
}
private final String name;
http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/JsonSerialization.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/JsonSerialization.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/JsonSerialization.java
new file mode 100644
index 0000000..15f4fef
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/util/JsonSerialization.java
@@ -0,0 +1,299 @@
+/*
+ * 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.util;
+
+import java.io.EOFException;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
+import com.fasterxml.jackson.core.JsonParseException;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+/**
+ * Support for marshalling objects to and from JSON.
+ *
+ * It constructs an object mapper as an instance field.
+ * and synchronizes access to those methods
+ * which use the mapper.
+ *
+ * This class was extracted from
+ * {@code org.apache.hadoop.registry.client.binding.JsonSerDeser},
+ * which is now a subclass of this class.
+ * @param <T> Type to marshal.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+public class JsonSerialization<T> {
+
+ private static final Logger LOG =
+ LoggerFactory.getLogger(JsonSerialization.class);
+ private static final String UTF_8 = "UTF-8";
+
+ private final Class<T> classType;
+ private final ObjectMapper mapper;
+
+ /**
+ * Create an instance bound to a specific type.
+ * @param classType class to marshall
+ * @param failOnUnknownProperties fail if an unknown property is encountered.
+ * @param pretty generate pretty (indented) output?
+ */
+ public JsonSerialization(Class<T> classType,
+ boolean failOnUnknownProperties, boolean pretty) {
+ Preconditions.checkArgument(classType != null, "null classType");
+ this.classType = classType;
+ this.mapper = new ObjectMapper();
+ mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES,
+ failOnUnknownProperties);
+ mapper.configure(SerializationFeature.INDENT_OUTPUT, pretty);
+ }
+
+ /**
+ * Get the simple name of the class type to be marshalled.
+ * @return the name of the class being marshalled
+ */
+ public String getName() {
+ return classType.getSimpleName();
+ }
+
+ /**
+ * Convert from JSON.
+ *
+ * @param json input
+ * @return the parsed JSON
+ * @throws IOException IO problems
+ * @throws JsonParseException If the input is not well-formatted
+ * @throws JsonMappingException failure to map from the JSON to this class
+ */
+ @SuppressWarnings("unchecked")
+ public synchronized T fromJson(String json)
+ throws IOException, JsonParseException, JsonMappingException {
+ if (json.isEmpty()) {
+ throw new EOFException("No data");
+ }
+ try {
+ return mapper.readValue(json, classType);
+ } catch (IOException e) {
+ LOG.error("Exception while parsing json : {}\n{}", e, json, e);
+ throw e;
+ }
+ }
+
+ /**
+ * Read from an input stream.
+ * @param stream stream to read from
+ * @return the parsed entity
+ * @throws IOException IO problems
+ * @throws JsonParseException If the input is not well-formatted
+ * @throws JsonMappingException failure to map from the JSON to this class
+ */
+ public synchronized T fromJsonStream(InputStream stream) throws IOException {
+ return mapper.readValue(stream, classType);
+ }
+
+ /**
+ * Load from a JSON text file.
+ * @param jsonFile input file
+ * @return the parsed JSON
+ * @throws IOException IO problems
+ * @throws JsonParseException If the input is not well-formatted
+ * @throws JsonMappingException failure to map from the JSON to this class
+ */
+ @SuppressWarnings("unchecked")
+ public synchronized T load(File jsonFile)
+ throws IOException, JsonParseException, JsonMappingException {
+ if (!jsonFile.isFile()) {
+ throw new FileNotFoundException("Not a file: " + jsonFile);
+ }
+ if (jsonFile.length() == 0) {
+ throw new EOFException("File is empty: " + jsonFile);
+ }
+ try {
+ return mapper.readValue(jsonFile, classType);
+ } catch (IOException e) {
+ LOG.error("Exception while parsing json file {}", jsonFile, e);
+ throw e;
+ }
+ }
+
+ /**
+ * Save to a local file. Any existing file is overwritten unless
+ * the OS blocks that.
+ * @param file file
+ * @param path path
+ * @throws IOException IO exception
+ */
+ public void save(File file, T instance) throws
+ IOException {
+ writeJsonAsBytes(instance, new FileOutputStream(file));
+ }
+
+ /**
+ * Convert from a JSON file.
+ * @param resource input file
+ * @return the parsed JSON
+ * @throws IOException IO problems
+ * @throws JsonParseException If the input is not well-formatted
+ * @throws JsonMappingException failure to map from the JSON to this class
+ */
+ @SuppressWarnings({"IOResourceOpenedButNotSafelyClosed"})
+ public synchronized T fromResource(String resource)
+ throws IOException, JsonParseException, JsonMappingException {
+ try (InputStream resStream = this.getClass()
+ .getResourceAsStream(resource)) {
+ if (resStream == null) {
+ throw new FileNotFoundException(resource);
+ }
+ return mapper.readValue(resStream, classType);
+ } catch (IOException e) {
+ LOG.error("Exception while parsing json resource {}", resource, e);
+ throw e;
+ }
+ }
+
+ /**
+ * clone by converting to JSON and back again.
+ * This is much less efficient than any Java clone process.
+ * @param instance instance to duplicate
+ * @return a new instance
+ * @throws IOException IO problems.
+ */
+ public T fromInstance(T instance) throws IOException {
+ return fromJson(toJson(instance));
+ }
+
+ /**
+ * Load from a Hadoop filesystem.
+ * There's a check for data availability after the file is open, by
+ * raising an EOFException if stream.available == 0.
+ * This allows for a meaningful exception without the round trip overhead
+ * of a getFileStatus call before opening the file. It may be brittle
+ * against an FS stream which doesn't return a value here, but the
+ * standard filesystems all do.
+ * JSON parsing and mapping problems
+ * are converted to IOEs.
+ * @param fs filesystem
+ * @param path path
+ * @return a loaded object
+ * @throws IOException IO or JSON parse problems
+ */
+ public T load(FileSystem fs, Path path) throws IOException {
+ try (FSDataInputStream dataInputStream = fs.open(path)) {
+ // throw an EOF exception if there is no data available.
+ if (dataInputStream.available() == 0) {
+ throw new EOFException("No data in " + path);
+ }
+ return fromJsonStream(dataInputStream);
+ } catch (JsonProcessingException e) {
+ throw new IOException(
+ String.format("Failed to read JSON file \"%s\": %s", path, e),
+ e);
+ }
+ }
+
+ /**
+ * Save to a Hadoop filesystem.
+ * @param fs filesystem
+ * @param path path
+ * @param overwrite should any existing file be overwritten
+ * @throws IOException IO exception
+ */
+ public void save(FileSystem fs, Path path, T instance,
+ boolean overwrite) throws
+ IOException {
+ writeJsonAsBytes(instance, fs.create(path, overwrite));
+ }
+
+ /**
+ * Write the JSON as bytes, then close the file.
+ * @param dataOutputStream an output stream that will always be closed
+ * @throws IOException on any failure
+ */
+ private void writeJsonAsBytes(T instance,
+ OutputStream dataOutputStream) throws IOException {
+ try {
+ dataOutputStream.write(toBytes(instance));
+ } finally {
+ dataOutputStream.close();
+ }
+ }
+
+ /**
+ * Convert JSON to bytes.
+ * @param instance instance to convert
+ * @return a byte array
+ * @throws IOException IO problems
+ */
+ public byte[] toBytes(T instance) throws IOException {
+ return mapper.writeValueAsBytes(instance);
+ }
+
+ /**
+ * Deserialize from a byte array.
+ * @param bytes byte array
+ * @throws IOException IO problems
+ * @throws EOFException not enough data
+ */
+ public T fromBytes(byte[] bytes) throws IOException {
+ return fromJson(new String(bytes, 0, bytes.length, UTF_8));
+ }
+
+ /**
+ * Convert an instance to a JSON string.
+ * @param instance instance to convert
+ * @return a JSON string description
+ * @throws JsonProcessingException Json generation problems
+ */
+ public synchronized String toJson(T instance) throws JsonProcessingException {
+ return mapper.writeValueAsString(instance);
+ }
+
+ /**
+ * Convert an instance to a string form for output. This is a robust
+ * operation which will convert any JSON-generating exceptions into
+ * error text.
+ * @param instance non-null instance
+ * @return a JSON string
+ */
+ public String toString(T instance) {
+ Preconditions.checkArgument(instance != null, "Null instance argument");
+ try {
+ return toJson(instance);
+ } catch (JsonProcessingException e) {
+ return "Failed to convert to a string: " + e;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml b/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
index 8db9f44..183faa5 100644
--- a/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
+++ b/hadoop-common-project/hadoop-common/src/main/resources/core-default.xml
@@ -1123,7 +1123,8 @@
<property>
<name>fs.s3a.multipart.purge.age</name>
<value>86400</value>
- <description>Minimum age in seconds of multipart uploads to purge.
+ <description>Minimum age in seconds of multipart uploads to purge
+ on startup if "fs.s3a.multipart.purge" is true
</description>
</property>
@@ -1345,6 +1346,120 @@
</property>
<property>
+ <name>fs.s3a.retry.limit</name>
+ <value>${fs.s3a.attempts.maximum}</value>
+ <description>
+ Number of times to retry any repeatable S3 client request on failure,
+ excluding throttling requests.
+ </description>
+</property>
+
+<property>
+ <name>fs.s3a.retry.interval</name>
+ <value>500ms</value>
+ <description>
+ Interval between attempts to retry operations for any reason other
+ than S3 throttle errors.
+ </description>
+</property>
+
+<property>
+ <name>fs.s3a.retry.throttle.limit</name>
+ <value>${fs.s3a.attempts.maximum}</value>
+ <description>
+ Number of times to retry any throttled request.
+ </description>
+</property>
+
+<property>
+ <name>fs.s3a.retry.throttle.interval</name>
+ <value>1000ms</value>
+ <description>
+ Interval between retry attempts on throttled requests.
+ </description>
+</property>
+
+<property>
+ <name>fs.s3a.committer.name</name>
+ <value>file</value>
+ <description>
+ Committer to create for output to S3A, one of:
+ "file", "directory", "partitioned", "magic".
+ </description>
+</property>
+
+<property>
+ <name>fs.s3a.committer.magic.enabled</name>
+ <value>false</value>
+ <description>
+ Enable support in the filesystem for the S3 "Magic" committer.
+ When working with AWS S3, S3Guard must be enabled for the destination
+ bucket, as consistent metadata listings are required.
+ </description>
+</property>
+
+<property>
+ <name>fs.s3a.committer.threads</name>
+ <value>8</value>
+ <description>
+ Number of threads in committers for parallel operations on files
+ (upload, commit, abort, delete...)
+ </description>
+</property>
+
+<property>
+ <name>fs.s3a.committer.staging.tmp.path</name>
+ <value>tmp/staging</value>
+ <description>
+ Path in the cluster filesystem for temporary data.
+ This is for HDFS, not the local filesystem.
+ It is only for the summary data of each file, not the actual
+ data being committed.
+ Using an unqualified path guarantees that the full path will be
+ generated relative to the home directory of the user creating the job,
+ hence private (assuming home directory permissions are secure).
+ </description>
+</property>
+
+<property>
+ <name>fs.s3a.committer.staging.unique-filenames</name>
+ <value>true</value>
+ <description>
+ Option for final files to have a unique name through job attempt info,
+ or the value of fs.s3a.committer.staging.uuid
+ When writing data with the "append" conflict option, this guarantees
+ that new data will not overwrite any existing data.
+ </description>
+</property>
+
+<property>
+ <name>fs.s3a.committer.staging.conflict-mode</name>
+ <value>fail</value>
+ <description>
+ Staging committer conflict resolution policy.
+ Supported: "fail", "append", "replace".
+ </description>
+</property>
+
+<property>
+ <name>fs.s3a.committer.staging.abort.pending.uploads</name>
+ <value>true</value>
+ <description>
+ Should the staging committers abort all pending uploads to the destination
+ directory?
+
+ Changing this if more than one partitioned committer is
+ writing to the same destination tree simultaneously; otherwise
+ the first job to complete will cancel all outstanding uploads from the
+ others. However, it may lead to leaked outstanding uploads from failed
+ tasks. If disabled, configure the bucket lifecycle to remove uploads
+ after a time period, and/or set up a workflow to explicitly delete
+ entries. Otherwise there is a risk that uncommitted uploads may run up
+ bills.
+ </description>
+</property>
+
+<property>
<name>fs.AbstractFileSystem.s3a.impl</name>
<value>org.apache.hadoop.fs.s3a.S3A</value>
<description>The implementation class of the S3A AbstractFileSystem.</description>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java
index e0cc7d6..d2cbca0 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/contract/ContractTestUtils.java
@@ -725,6 +725,8 @@ public class ContractTestUtils extends Assert {
/**
* Read in "length" bytes, convert to an ascii string.
+ * This uses {@link #toChar(byte)} to escape bytes, so cannot be used
+ * for round trip operations.
* @param fs filesystem
* @param path path to read
* @param length #of bytes to read.
@@ -742,6 +744,28 @@ public class ContractTestUtils extends Assert {
}
/**
+ * Read in "length" bytes, convert to UTF8 string.
+ * @param fs filesystem
+ * @param path path to read
+ * @param length #of bytes to read. If -1: use file length.
+ * @return the bytes read and converted to a string
+ * @throws IOException IO problems
+ */
+ public static String readUTF8(FileSystem fs,
+ Path path,
+ int length) throws IOException {
+ if (length < 0) {
+ FileStatus status = fs.getFileStatus(path);
+ length = (int) status.getLen();
+ }
+ try (FSDataInputStream in = fs.open(path)) {
+ byte[] buf = new byte[length];
+ in.readFully(0, buf);
+ return new String(buf, "UTF-8");
+ }
+ }
+
+ /**
* Take an array of filestats and convert to a string
* (prefixed with/ a [%02d] counter).
* @param stats array of stats
@@ -857,11 +881,30 @@ public class ContractTestUtils extends Assert {
*/
public static void assertPathExists(FileSystem fileSystem, String message,
Path path) throws IOException {
- if (!fileSystem.exists(path)) {
+ verifyPathExists(fileSystem, message, path);
+ }
+
+ /**
+ * Verify that a path exists, returning the file status of the path.
+ *
+ * @param fileSystem filesystem to examine
+ * @param message message to include in the assertion failure message
+ * @param path path in the filesystem
+ * @throws FileNotFoundException raised if the path is missing
+ * @throws IOException IO problems
+ */
+ public static FileStatus verifyPathExists(FileSystem fileSystem,
+ String message,
+ Path path) throws IOException {
+ try {
+ return fileSystem.getFileStatus(path);
+ } catch (FileNotFoundException e) {
//failure, report it
- ls(fileSystem, path.getParent());
- throw new FileNotFoundException(message + ": not found " + path
- + " in " + path.getParent());
+ LOG.error("{}: not found {}; parent listing is:\n{}",
+ message, path, ls(fileSystem, path.getParent()));
+ throw (IOException)new FileNotFoundException(
+ message + ": not found " + path + " in " + path.getParent())
+ .initCause(e);
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/test/GenericTestUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/test/GenericTestUtils.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/test/GenericTestUtils.java
index 4cb9f8b..0db6c73 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/test/GenericTestUtils.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/test/GenericTestUtils.java
@@ -316,20 +316,37 @@ public abstract class GenericTestUtils {
/**
* Assert that an exception's <code>toString()</code> value
* contained the expected text.
- * @param string expected string
+ * @param expectedText expected string
* @param t thrown exception
* @throws AssertionError if the expected string is not found
*/
- public static void assertExceptionContains(String string, Throwable t) {
+ public static void assertExceptionContains(String expectedText, Throwable t) {
+ assertExceptionContains(expectedText, t, "");
+ }
+
+ /**
+ * Assert that an exception's <code>toString()</code> value
+ * contained the expected text.
+ * @param expectedText expected string
+ * @param t thrown exception
+ * @param message any extra text for the string
+ * @throws AssertionError if the expected string is not found
+ */
+ public static void assertExceptionContains(String expectedText,
+ Throwable t,
+ String message) {
Assert.assertNotNull(E_NULL_THROWABLE, t);
String msg = t.toString();
if (msg == null) {
throw new AssertionError(E_NULL_THROWABLE_STRING, t);
}
- if (!msg.contains(string)) {
- throw new AssertionError("Expected to find '" + string + "' "
- + E_UNEXPECTED_EXCEPTION + ":"
- + StringUtils.stringifyException(t),
+ if (expectedText != null && !msg.contains(expectedText)) {
+ String prefix = org.apache.commons.lang.StringUtils.isEmpty(message)
+ ? "" : (message + ": ");
+ throw new AssertionError(
+ String.format("%s Expected to find '%s' %s: %s",
+ prefix, expectedText, E_UNEXPECTED_EXCEPTION,
+ StringUtils.stringifyException(t)),
t);
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/test/HadoopTestBase.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/test/HadoopTestBase.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/test/HadoopTestBase.java
index 43d5be8..cb7df4b 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/test/HadoopTestBase.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/test/HadoopTestBase.java
@@ -17,29 +17,37 @@
*/
package org.apache.hadoop.test;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
import org.junit.Rule;
+import org.junit.rules.TestName;
import org.junit.rules.Timeout;
/**
* A base class for JUnit4 tests that sets a default timeout for all tests
- * that subclass this test
+ * that subclass this test.
+ *
+ * Threads are named to the method being executed, for ease of diagnostics
+ * in logs and thread dumps.
*/
-public abstract class HadoopTestBase {
+public abstract class HadoopTestBase extends Assert {
+
/**
- * System property name to set the test timeout: {@value}
+ * System property name to set the test timeout: {@value}.
*/
public static final String PROPERTY_TEST_DEFAULT_TIMEOUT =
- "test.default.timeout";
+ "test.default.timeout";
/**
* The default timeout (in milliseconds) if the system property
* {@link #PROPERTY_TEST_DEFAULT_TIMEOUT}
- * is not set: {@value}
+ * is not set: {@value}.
*/
public static final int TEST_DEFAULT_TIMEOUT_VALUE = 100000;
/**
- * The JUnit rule that sets the default timeout for tests
+ * The JUnit rule that sets the default timeout for tests.
*/
@Rule
public Timeout defaultTimeout = retrieveTestTimeout();
@@ -64,4 +72,35 @@ public abstract class HadoopTestBase {
}
return new Timeout(millis);
}
+
+ /**
+ * The method name.
+ */
+ @Rule
+ public TestName methodName = new TestName();
+
+ /**
+ * Get the method name; defaults to the value of {@link #methodName}.
+ * Subclasses may wish to override it, which will tune the thread naming.
+ * @return the name of the method.
+ */
+ protected String getMethodName() {
+ return methodName.getMethodName();
+ }
+
+ /**
+ * Static initializer names this thread "JUnit".
+ */
+ @BeforeClass
+ public static void nameTestThread() {
+ Thread.currentThread().setName("JUnit");
+ }
+
+ /**
+ * Before each method, the thread is renamed to match the method name.
+ */
+ @Before
+ public void nameThreadToMethod() {
+ Thread.currentThread().setName("JUnit-" + getMethodName());
+ }
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/test/LambdaTestUtils.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/test/LambdaTestUtils.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/test/LambdaTestUtils.java
index 3ea9ab8..22208f7 100644
--- a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/test/LambdaTestUtils.java
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/test/LambdaTestUtils.java
@@ -19,11 +19,13 @@
package org.apache.hadoop.test;
import com.google.common.base.Preconditions;
+import org.junit.Assert;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.util.Time;
+import java.util.Optional;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeoutException;
@@ -43,7 +45,7 @@ import java.util.concurrent.TimeoutException;
* check; useful when checking the contents of the exception.
*/
public final class LambdaTestUtils {
- public static final Logger LOG =
+ private static final Logger LOG =
LoggerFactory.getLogger(LambdaTestUtils.class);
private LambdaTestUtils() {
@@ -60,6 +62,7 @@ public final class LambdaTestUtils {
* Interface to implement for converting a timeout into some form
* of exception to raise.
*/
+ @FunctionalInterface
public interface TimeoutHandler {
/**
@@ -371,16 +374,11 @@ public final class LambdaTestUtils {
Class<E> clazz,
Callable<T> eval)
throws Exception {
- try {
- T result = eval.call();
- throw new AssertionError("Expected an exception, got "
- + robustToString(result));
- } catch (Throwable e) {
- if (clazz.isAssignableFrom(e.getClass())) {
- return (E)e;
- }
- throw e;
- }
+ return intercept(clazz,
+ null,
+ "Expected a " + clazz.getName() + " to be thrown," +
+ " but got the result: ",
+ eval);
}
/**
@@ -451,6 +449,59 @@ public final class LambdaTestUtils {
}
/**
+ * Intercept an exception; throw an {@code AssertionError} if one not raised.
+ * The caught exception is rethrown if it is of the wrong class or
+ * does not contain the text defined in {@code contained}.
+ * <p>
+ * Example: expect deleting a nonexistent file to raise a
+ * {@code FileNotFoundException} with the {@code toString()} value
+ * containing the text {@code "missing"}.
+ * <pre>
+ * FileNotFoundException ioe = intercept(FileNotFoundException.class,
+ * "missing",
+ * "path should not be found",
+ * () -> {
+ * filesystem.delete(new Path("/missing"), false);
+ * });
+ * </pre>
+ *
+ * @param clazz class of exception; the raised exception must be this class
+ * <i>or a subclass</i>.
+ * @param contained string which must be in the {@code toString()} value
+ * of the exception
+ * @param message any message tho include in exception/log messages
+ * @param eval expression to eval
+ * @param <T> return type of expression
+ * @param <E> exception class
+ * @return the caught exception if it was of the expected type and contents
+ * @throws Exception any other exception raised
+ * @throws AssertionError if the evaluation call didn't raise an exception.
+ * The error includes the {@code toString()} value of the result, if this
+ * can be determined.
+ * @see GenericTestUtils#assertExceptionContains(String, Throwable)
+ */
+ public static <T, E extends Throwable> E intercept(
+ Class<E> clazz,
+ String contained,
+ String message,
+ Callable<T> eval)
+ throws Exception {
+ E ex;
+ try {
+ T result = eval.call();
+ throw new AssertionError(message + ": " + robustToString(result));
+ } catch (Throwable e) {
+ if (!clazz.isAssignableFrom(e.getClass())) {
+ throw e;
+ } else {
+ ex = (E) e;
+ }
+ }
+ GenericTestUtils.assertExceptionContains(contained, ex, message);
+ return ex;
+ }
+
+ /**
* Variant of {@link #intercept(Class, Callable)} to simplify void
* invocations.
* @param clazz class of exception; the raised exception must be this class
@@ -468,9 +519,41 @@ public final class LambdaTestUtils {
String contained,
VoidCallable eval)
throws Exception {
- E ex = intercept(clazz, eval);
- GenericTestUtils.assertExceptionContains(contained, ex);
- return ex;
+ return intercept(clazz, contained,
+ "Expecting " + clazz.getName()
+ + (contained != null? (" with text " + contained) : "")
+ + " but got ",
+ () -> {
+ eval.call();
+ return "void";
+ });
+ }
+
+ /**
+ * Variant of {@link #intercept(Class, Callable)} to simplify void
+ * invocations.
+ * @param clazz class of exception; the raised exception must be this class
+ * <i>or a subclass</i>.
+ * @param contained string which must be in the {@code toString()} value
+ * of the exception
+ * @param message any message tho include in exception/log messages
+ * @param eval expression to eval
+ * @param <E> exception class
+ * @return the caught exception if it was of the expected type
+ * @throws Exception any other exception raised
+ * @throws AssertionError if the evaluation call didn't raise an exception.
+ */
+ public static <E extends Throwable> E intercept(
+ Class<E> clazz,
+ String contained,
+ String message,
+ VoidCallable eval)
+ throws Exception {
+ return intercept(clazz, contained, message,
+ () -> {
+ eval.call();
+ return "void";
+ });
}
/**
@@ -495,6 +578,38 @@ public final class LambdaTestUtils {
}
/**
+ * Assert that an optional value matches an expected one;
+ * checks include null and empty on the actual value.
+ * @param message message text
+ * @param expected expected value
+ * @param actual actual optional value
+ * @param <T> type
+ */
+ public static <T> void assertOptionalEquals(String message,
+ T expected,
+ Optional<T> actual) {
+ Assert.assertNotNull(message, actual);
+ Assert.assertTrue(message +" -not present", actual.isPresent());
+ Assert.assertEquals(message, expected, actual.get());
+ }
+
+ /**
+ * Assert that an optional value matches an expected one;
+ * checks include null and empty on the actual value.
+ * @param message message text
+ * @param expected expected value
+ * @param actual actual optional value
+ * @param <T> type
+ */
+ public static <T> void assertOptionalUnset(String message,
+ Optional<T> actual) {
+ Assert.assertNotNull(message, actual);
+ if (actual.isPresent()) {
+ Assert.fail("Expected empty option, got " + actual.get().toString());
+ }
+ }
+
+ /**
* Returns {@code TimeoutException} on a timeout. If
* there was a inner class passed in, includes it as the
* inner failure.
@@ -638,6 +753,7 @@ public final class LambdaTestUtils {
* A simple interface for lambdas, which returns nothing; this exists
* to simplify lambda tests on operations with no return value.
*/
+ @FunctionalInterface
public interface VoidCallable {
void call() throws Exception;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestJsonSerialization.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestJsonSerialization.java b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestJsonSerialization.java
new file mode 100644
index 0000000..991697d
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/util/TestJsonSerialization.java
@@ -0,0 +1,185 @@
+/*
+ * 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.util;
+
+import java.io.EOFException;
+import java.io.File;
+import java.io.FileNotFoundException;
+import java.io.Serializable;
+import java.util.Objects;
+
+import com.fasterxml.jackson.core.JsonParseException;
+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 org.apache.hadoop.test.LambdaTestUtils;
+
+/**
+ * Test the JSON serialization helper.
+ */
+public class TestJsonSerialization extends HadoopTestBase {
+
+ private final JsonSerialization<KeyVal> serDeser =
+ new JsonSerialization<>(KeyVal.class, true, true);
+
+ private final KeyVal source = new KeyVal("key", "1");
+
+ private static class KeyVal implements Serializable {
+ private String name;
+ private String value;
+
+ KeyVal(String name, String value) {
+ this.name = name;
+ this.value = value;
+ }
+
+ KeyVal() {
+ }
+
+ @Override
+ public String toString() {
+ final StringBuilder sb = new StringBuilder("SimpleJson{");
+ sb.append("name='").append(name).append('\'');
+ sb.append(", value='").append(value).append('\'');
+ sb.append('}');
+ return sb.toString();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ KeyVal that = (KeyVal) o;
+ return Objects.equals(name, that.name) &&
+ Objects.equals(value, that.value);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(name, value);
+ }
+
+ public String getName() {
+ return name;
+ }
+
+ public void setName(String name) {
+ this.name = name;
+ }
+
+ public String getValue() {
+ return value;
+ }
+
+ public void setValue(String value) {
+ this.value = value;
+ }
+ }
+
+ @Test
+ public void testStringRoundTrip() throws Throwable {
+ String wire = serDeser.toJson(source);
+ KeyVal unmarshalled = serDeser.fromJson(wire);
+ assertEquals("Failed to unmarshall: " + wire, source, unmarshalled);
+ }
+
+ @Test
+ public void testBytesRoundTrip() throws Throwable {
+ byte[] wire = serDeser.toBytes(source);
+ KeyVal unmarshalled = serDeser.fromBytes(wire);
+ assertEquals(source, unmarshalled);
+ }
+
+ @Test
+ public void testBadBytesRoundTrip() throws Throwable {
+ LambdaTestUtils.intercept(JsonParseException.class,
+ "token",
+ () -> serDeser.fromBytes(new byte[]{'a'}));
+ }
+
+ @Test
+ public void testCloneViaJson() throws Throwable {
+ KeyVal unmarshalled = serDeser.fromInstance(source);
+ assertEquals(source, unmarshalled);
+ }
+
+ @Test
+ public void testFileRoundTrip() throws Throwable {
+ File tempFile = File.createTempFile("Keyval", ".json");
+ tempFile.delete();
+ try {
+ serDeser.save(tempFile, source);
+ assertEquals(source, serDeser.load(tempFile));
+ } finally {
+ tempFile.delete();
+ }
+ }
+
+ @Test
+ public void testEmptyFile() throws Throwable {
+ File tempFile = File.createTempFile("Keyval", ".json");
+ try {
+ LambdaTestUtils.intercept(EOFException.class,
+ "empty",
+ () -> serDeser.load(tempFile));
+ } finally {
+ tempFile.delete();
+ }
+ }
+
+ @Test
+ public void testFileSystemRoundTrip() throws Throwable {
+ File tempFile = File.createTempFile("Keyval", ".json");
+ tempFile.delete();
+ Path tempPath = new Path(tempFile.toURI());
+ LocalFileSystem fs = FileSystem.getLocal(new Configuration());
+ try {
+ serDeser.save(fs, tempPath, source, false);
+ assertEquals(source, serDeser.load(fs, tempPath));
+ } finally {
+ fs.delete(tempPath, false);
+ }
+ }
+
+ @Test
+ public void testFileSystemEmptyPath() throws Throwable {
+ File tempFile = File.createTempFile("Keyval", ".json");
+ Path tempPath = new Path(tempFile.toURI());
+ LocalFileSystem fs = FileSystem.getLocal(new Configuration());
+ try {
+ LambdaTestUtils.intercept(EOFException.class,
+ () -> serDeser.load(fs, tempPath));
+ fs.delete(tempPath, false);
+ LambdaTestUtils.intercept(FileNotFoundException.class,
+ () -> serDeser.load(fs, tempPath));
+ } finally {
+ fs.delete(tempPath, false);
+ }
+ }
+
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/TestMapreduceConfigFields.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/TestMapreduceConfigFields.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/TestMapreduceConfigFields.java
index 5d42fbf..f8aaab7 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/TestMapreduceConfigFields.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/TestMapreduceConfigFields.java
@@ -27,6 +27,7 @@ import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.NLineInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputCommitter;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
+import org.apache.hadoop.mapreduce.lib.output.PathOutputCommitterFactory;
import org.apache.hadoop.mapreduce.v2.jobhistory.JHAdminConfig;
/**
@@ -53,14 +54,23 @@ public class TestMapreduceConfigFields extends TestConfigurationFieldsBase {
@SuppressWarnings("deprecation")
@Override
public void initializeMemberVariables() {
- xmlFilename = new String("mapred-default.xml");
- configurationClasses = new Class[] { MRJobConfig.class, MRConfig.class,
- JHAdminConfig.class, ShuffleHandler.class, FileOutputFormat.class,
- FileInputFormat.class, Job.class, NLineInputFormat.class,
- JobConf.class, FileOutputCommitter.class };
+ xmlFilename = "mapred-default.xml";
+ configurationClasses = new Class[] {
+ MRJobConfig.class,
+ MRConfig.class,
+ JHAdminConfig.class,
+ ShuffleHandler.class,
+ FileOutputFormat.class,
+ FileInputFormat.class,
+ Job.class,
+ NLineInputFormat.class,
+ JobConf.class,
+ FileOutputCommitter.class,
+ PathOutputCommitterFactory.class
+ };
// Initialize used variables
- configurationPropsToSkipCompare = new HashSet<String>();
+ configurationPropsToSkipCompare = new HashSet<>();
// Set error modes
errorIfMissingConfigProps = true;
@@ -82,6 +92,11 @@ public class TestMapreduceConfigFields extends TestConfigurationFieldsBase {
MRJobConfig.MAP_RESOURCE_TYPE_PREFIX);
configurationPropsToSkipCompare.add(
MRJobConfig.REDUCE_RESOURCE_TYPE_PREFIX);
+
+ // PathOutputCommitterFactory values
+ xmlPrefixToSkipCompare = new HashSet<>();
+ xmlPrefixToSkipCompare.add(
+ PathOutputCommitterFactory.COMMITTER_FACTORY_SCHEME);
}
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/BindingPathOutputCommitter.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/BindingPathOutputCommitter.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/BindingPathOutputCommitter.java
new file mode 100644
index 0000000..f12678b
--- /dev/null
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/BindingPathOutputCommitter.java
@@ -0,0 +1,184 @@
+/*
+ * 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.mapreduce.lib.output;
+
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapreduce.JobContext;
+import org.apache.hadoop.mapreduce.JobStatus;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+
+/**
+ * This is a special committer which creates the factory for the committer and
+ * runs off that. Why does it exist? So that you can explicitly instantiate
+ * a committer by classname and yet still have the actual implementation
+ * driven dynamically by the factory options and destination filesystem.
+ * This simplifies integration
+ * with existing code which takes the classname of a committer.
+ * There's no factory for this, as that would lead to a loop.
+ *
+ * All commit protocol methods and accessors are delegated to the
+ * wrapped committer.
+ *
+ * How to use:
+ *
+ * <ol>
+ * <li>
+ * In applications which take a classname of committer in
+ * a configuration option, set it to the canonical name of this class
+ * (see {@link #NAME}). When this class is instantiated, it will
+ * use the factory mechanism to locate the configured committer for the
+ * destination.
+ * </li>
+ * <li>
+ * In code, explicitly create an instance of this committer through
+ * its constructor, then invoke commit lifecycle operations on it.
+ * The dynamically configured committer will be created in the constructor
+ * and have the lifecycle operations relayed to it.
+ * </li>
+ * </ol>
+ *
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Unstable
+public class BindingPathOutputCommitter extends PathOutputCommitter {
+
+ /**
+ * The classname for use in configurations.
+ */
+ public static final String NAME
+ = BindingPathOutputCommitter.class.getCanonicalName();
+
+ /**
+ * The bound committer.
+ */
+ private final PathOutputCommitter committer;
+
+ /**
+ * Instantiate.
+ * @param outputPath output path (may be null)
+ * @param context task context
+ * @throws IOException on any failure.
+ */
+ public BindingPathOutputCommitter(Path outputPath,
+ TaskAttemptContext context) throws IOException {
+ super(outputPath, context);
+ committer = PathOutputCommitterFactory.getCommitterFactory(outputPath,
+ context.getConfiguration())
+ .createOutputCommitter(outputPath, context);
+ }
+
+ @Override
+ public Path getOutputPath() {
+ return committer.getOutputPath();
+ }
+
+ @Override
+ public Path getWorkPath() throws IOException {
+ return committer.getWorkPath();
+ }
+
+ @Override
+ public void setupJob(JobContext jobContext) throws IOException {
+ committer.setupJob(jobContext);
+ }
+
+ @Override
+ public void setupTask(TaskAttemptContext taskContext) throws IOException {
+ committer.setupTask(taskContext);
+ }
+
+ @Override
+ public boolean needsTaskCommit(TaskAttemptContext taskContext)
+ throws IOException {
+ return committer.needsTaskCommit(taskContext);
+ }
+
+ @Override
+ public void commitTask(TaskAttemptContext taskContext) throws IOException {
+ committer.commitTask(taskContext);
+ }
+
+ @Override
+ public void abortTask(TaskAttemptContext taskContext) throws IOException {
+ committer.abortTask(taskContext);
+ }
+
+ @Override
+ @SuppressWarnings("deprecation")
+ public void cleanupJob(JobContext jobContext) throws IOException {
+ super.cleanupJob(jobContext);
+ }
+
+ @Override
+ public void commitJob(JobContext jobContext) throws IOException {
+ committer.commitJob(jobContext);
+ }
+
+ @Override
+ public void abortJob(JobContext jobContext, JobStatus.State state)
+ throws IOException {
+ committer.abortJob(jobContext, state);
+ }
+
+ @SuppressWarnings("deprecation")
+ @Override
+ public boolean isRecoverySupported() {
+ return committer.isRecoverySupported();
+ }
+
+ @Override
+ public boolean isCommitJobRepeatable(JobContext jobContext)
+ throws IOException {
+ return committer.isCommitJobRepeatable(jobContext);
+ }
+
+ @Override
+ public boolean isRecoverySupported(JobContext jobContext) throws IOException {
+ return committer.isRecoverySupported(jobContext);
+ }
+
+ @Override
+ public void recoverTask(TaskAttemptContext taskContext) throws IOException {
+ committer.recoverTask(taskContext);
+ }
+
+ @Override
+ public boolean hasOutputPath() {
+ return committer.hasOutputPath();
+ }
+
+ @Override
+ public String toString() {
+ return "BindingPathOutputCommitter{"
+ + "committer=" + committer +
+ '}';
+ }
+
+ /**
+ * Get the inner committer.
+ * @return the bonded committer.
+ */
+ public PathOutputCommitter getCommitter() {
+ return committer;
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/FileOutputCommitter.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/FileOutputCommitter.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/FileOutputCommitter.java
index 325b2e7..86af2cf 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/FileOutputCommitter.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/FileOutputCommitter.java
@@ -155,17 +155,11 @@ public class FileOutputCommitter extends PathOutputCommitter {
* @return the path where final output of the job should be placed. This
* could also be considered the committed application attempt path.
*/
- private Path getOutputPath() {
+ @Override
+ public Path getOutputPath() {
return this.outputPath;
}
-
- /**
- * @return true if we have an output path set, else false.
- */
- private boolean hasOutputPath() {
- return this.outputPath != null;
- }
-
+
/**
* @return the path where the output of pending job attempts are
* stored.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/FileOutputCommitterFactory.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/FileOutputCommitterFactory.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/FileOutputCommitterFactory.java
new file mode 100644
index 0000000..12b2841
--- /dev/null
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/FileOutputCommitterFactory.java
@@ -0,0 +1,38 @@
+/*
+ * 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.mapreduce.lib.output;
+
+import java.io.IOException;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+
+/**
+ * Creates a {@link FileOutputCommitter}, always.
+ */
+public final class FileOutputCommitterFactory
+ extends PathOutputCommitterFactory {
+
+ @Override
+ public PathOutputCommitter createOutputCommitter(Path outputPath,
+ TaskAttemptContext context) throws IOException {
+ return createFileOutputCommitter(outputPath, context);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/FileOutputFormat.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/FileOutputFormat.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/FileOutputFormat.java
index 0e7efa3..bbda26a 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/FileOutputFormat.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/FileOutputFormat.java
@@ -328,12 +328,14 @@ public abstract class FileOutputFormat<K, V> extends OutputFormat<K, V> {
job.getConfiguration().set(BASE_OUTPUT_NAME, name);
}
- public synchronized
- OutputCommitter getOutputCommitter(TaskAttemptContext context
- ) throws IOException {
+ public synchronized
+ OutputCommitter getOutputCommitter(TaskAttemptContext context)
+ throws IOException {
if (committer == null) {
Path output = getOutputPath(context);
- committer = new FileOutputCommitter(output, context);
+ committer = PathOutputCommitterFactory.getCommitterFactory(
+ output,
+ context.getConfiguration()).createOutputCommitter(output, context);
}
return committer;
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/NamedCommitterFactory.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/NamedCommitterFactory.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/NamedCommitterFactory.java
new file mode 100644
index 0000000..b7378af
--- /dev/null
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/NamedCommitterFactory.java
@@ -0,0 +1,79 @@
+/*
+ * 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.mapreduce.lib.output;
+
+import java.io.IOException;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
+
+import com.google.common.base.Preconditions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapreduce.JobContext;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+
+/**
+ * A factory which creates any named committer identified
+ * in the option {@link PathOutputCommitterFactory#NAMED_COMMITTER_CLASS}.
+ */
+public final class NamedCommitterFactory extends
+ PathOutputCommitterFactory {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(NamedCommitterFactory.class);
+
+ @SuppressWarnings("JavaReflectionMemberAccess")
+ @Override
+ public PathOutputCommitter createOutputCommitter(Path outputPath,
+ TaskAttemptContext context) throws IOException {
+ Class<? extends PathOutputCommitter> clazz = loadCommitterClass(context);
+ LOG.debug("Using PathOutputCommitter implementation {}", clazz);
+ try {
+ Constructor<? extends PathOutputCommitter> ctor
+ = clazz.getConstructor(Path.class, TaskAttemptContext.class);
+ return ctor.newInstance(outputPath, context);
+ } catch (NoSuchMethodException
+ | InstantiationException
+ | IllegalAccessException
+ | InvocationTargetException e) {
+ throw new IOException("Failed to create " + clazz
+ + ":" + e, e);
+ }
+ }
+
+ /**
+ * Load the class named in {@link #NAMED_COMMITTER_CLASS}.
+ * @param context job or task context
+ * @return the committer class
+ * @throws IOException if no committer was defined.
+ */
+ private Class<? extends PathOutputCommitter> loadCommitterClass(
+ JobContext context) throws IOException {
+ Preconditions.checkNotNull(context, "null context");
+ Configuration conf = context.getConfiguration();
+ String value = conf.get(NAMED_COMMITTER_CLASS, "");
+ if (value.isEmpty()) {
+ throw new IOException("No committer defined in " + NAMED_COMMITTER_CLASS);
+ }
+ return conf.getClass(NAMED_COMMITTER_CLASS,
+ FileOutputCommitter.class, PathOutputCommitter.class);
+ }
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/PathOutputCommitter.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/PathOutputCommitter.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/PathOutputCommitter.java
index 2df30ba..3679d9f 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/PathOutputCommitter.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/PathOutputCommitter.java
@@ -76,9 +76,26 @@ public abstract class PathOutputCommitter extends OutputCommitter {
}
/**
+ * Get the final directory where work will be placed once the job
+ * is committed. This may be null, in which case, there is no output
+ * path to write data to.
+ * @return the path where final output of the job should be placed.
+ */
+ public abstract Path getOutputPath();
+
+ /**
+ * Predicate: is there an output path?
+ * @return true if we have an output path set, else false.
+ */
+ public boolean hasOutputPath() {
+ return getOutputPath() != null;
+ }
+
+ /**
* Get the directory that the task should write results into.
* Warning: there's no guarantee that this work path is on the same
* FS as the final output, or that it's visible across machines.
+ * May be null.
* @return the work directory
* @throws IOException IO problem
*/
http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/PathOutputCommitterFactory.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/PathOutputCommitterFactory.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/PathOutputCommitterFactory.java
new file mode 100644
index 0000000..0df14d1
--- /dev/null
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/output/PathOutputCommitterFactory.java
@@ -0,0 +1,204 @@
+/*
+ * 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.mapreduce.lib.output;
+
+import java.io.IOException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.commons.lang.StringUtils;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.conf.Configured;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.util.ReflectionUtils;
+
+/**
+ * A factory for committers implementing the {@link PathOutputCommitter}
+ * methods, and so can be used from {@link FileOutputFormat}.
+ * The base implementation returns {@link FileOutputCommitter} instances.
+ *
+ * Algorithm:
+ * <ol>
+ * <ul>If an explicit committer factory is named, it is used.</ul>
+ * <ul>The output path is examined.
+ * If is non null and there is an explicit schema for that filesystem,
+ * its factory is instantiated.</ul>
+ * <ul>Otherwise, an instance of {@link FileOutputCommitter} is
+ * created.</ul>
+ * </ol>
+ *
+ * In {@link FileOutputFormat}, the created factory has its method
+ * {@link #createOutputCommitter(Path, TaskAttemptContext)} with a task
+ * attempt context and a possibly null path.
+ *
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Unstable
+public class PathOutputCommitterFactory extends Configured {
+ private static final Logger LOG =
+ LoggerFactory.getLogger(PathOutputCommitterFactory.class);
+
+ /**
+ * Name of the configuration option used to configure the
+ * output committer factory to use unless there is a specific
+ * one for a schema.
+ */
+ public static final String COMMITTER_FACTORY_CLASS =
+ "mapreduce.outputcommitter.factory.class";
+
+ /**
+ * Scheme prefix for per-filesystem scheme committers.
+ */
+ public static final String COMMITTER_FACTORY_SCHEME =
+ "mapreduce.outputcommitter.factory.scheme";
+
+ /**
+ * String format pattern for per-filesystem scheme committers.
+ */
+ public static final String COMMITTER_FACTORY_SCHEME_PATTERN =
+ COMMITTER_FACTORY_SCHEME + ".%s";
+
+
+ /**
+ * The {@link FileOutputCommitter} factory.
+ */
+ public static final String FILE_COMMITTER_FACTORY =
+ "org.apache.hadoop.mapreduce.lib.output.FileOutputCommitterFactory";
+
+ /**
+ * The {@link FileOutputCommitter} factory.
+ */
+ public static final String NAMED_COMMITTER_FACTORY =
+ "org.apache.hadoop.mapreduce.lib.output.NamedCommitterFactory";
+
+ /**
+ * The named output committer.
+ * Creates any committer listed in
+ */
+ public static final String NAMED_COMMITTER_CLASS =
+ "mapreduce.outputcommitter.named.classname";
+
+ /**
+ * Default committer factory name: {@value}.
+ */
+ public static final String COMMITTER_FACTORY_DEFAULT =
+ FILE_COMMITTER_FACTORY;
+
+ /**
+ * Create an output committer for a task attempt.
+ * @param outputPath output path. This may be null.
+ * @param context context
+ * @return a new committer
+ * @throws IOException problems instantiating the committer
+ */
+ public PathOutputCommitter createOutputCommitter(
+ Path outputPath,
+ TaskAttemptContext context) throws IOException {
+ return createFileOutputCommitter(outputPath, context);
+ }
+
+ /**
+ * Create an instance of the default committer, a {@link FileOutputCommitter}
+ * for a task.
+ * @param outputPath the task's output path, or or null if no output path
+ * has been defined.
+ * @param context the task attempt context
+ * @return the committer to use
+ * @throws IOException problems instantiating the committer
+ */
+ protected final PathOutputCommitter createFileOutputCommitter(
+ Path outputPath,
+ TaskAttemptContext context) throws IOException {
+ LOG.debug("Creating FileOutputCommitter for path {} and context {}",
+ outputPath, context);
+ return new FileOutputCommitter(outputPath, context);
+ }
+
+ /**
+ * Get the committer factory for a configuration.
+ * @param outputPath the job's output path. If null, it means that the
+ * schema is unknown and a per-schema factory cannot be determined.
+ * @param conf configuration
+ * @return an instantiated committer factory
+ */
+ public static PathOutputCommitterFactory getCommitterFactory(
+ Path outputPath,
+ Configuration conf) {
+ // determine which key to look up the overall one or a schema-specific
+ // key
+ LOG.debug("Looking for committer factory for path {}", outputPath);
+ String key = COMMITTER_FACTORY_CLASS;
+ if (StringUtils.isEmpty(conf.getTrimmed(key)) && outputPath != null) {
+ // there is no explicit factory and there's an output path
+ // Get the scheme of the destination
+ String scheme = outputPath.toUri().getScheme();
+
+ // and see if it has a key
+ String schemeKey = String.format(COMMITTER_FACTORY_SCHEME_PATTERN,
+ scheme);
+ if (StringUtils.isNotEmpty(conf.getTrimmed(schemeKey))) {
+ // it does, so use that key in the classname lookup
+ LOG.debug("Using schema-specific factory for {}", outputPath);
+ key = schemeKey;
+ } else {
+ LOG.debug("No scheme-specific factory defined in {}", schemeKey);
+ }
+ }
+
+ // create the factory. Before using Configuration.getClass, check
+ // for an empty configuration value, as that raises ClassNotFoundException.
+ Class<? extends PathOutputCommitterFactory> factory;
+ String trimmedValue = conf.getTrimmed(key, "");
+ if (StringUtils.isEmpty(trimmedValue)) {
+ // empty/null value, use default
+ LOG.debug("No output committer factory defined,"
+ + " defaulting to FileOutputCommitterFactory");
+ factory = FileOutputCommitterFactory.class;
+ } else {
+ // key is set, get the class
+ factory = conf.getClass(key,
+ FileOutputCommitterFactory.class,
+ PathOutputCommitterFactory.class);
+ LOG.debug("Using OutputCommitter factory class {} from key {}",
+ factory, key);
+ }
+ return ReflectionUtils.newInstance(factory, conf);
+ }
+
+ /**
+ * Create the committer factory for a task attempt & destination, then
+ * create the committer from it.
+ * @param outputPath the task's output path, or or null if no output path
+ * has been defined.
+ * @param context the task attempt context
+ * @return the committer to use
+ * @throws IOException problems instantiating the committer
+ */
+ public static PathOutputCommitter createCommitter(Path outputPath,
+ TaskAttemptContext context) throws IOException {
+ return getCommitterFactory(outputPath,
+ context.getConfiguration())
+ .createOutputCommitter(outputPath, context);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml
index 9d166c7..1e432ce 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/resources/mapred-default.xml
@@ -2043,4 +2043,26 @@
<name>mapreduce.job.send-token-conf</name>
<value></value>
</property>
+
+<property>
+ <description>
+ The name of an output committer factory for MRv2 FileOutputFormat to use
+ for committing work. If set, overrides any per-filesystem committer
+ defined for the destination filesystem.
+ </description>
+ <name>mapreduce.outputcommitter.factory.class</name>
+ <value></value>
+</property>
+
+
+<property>
+ <name>mapreduce.outputcommitter.factory.scheme.s3a</name>
+ <value>org.apache.hadoop.fs.s3a.commit.S3ACommitterFactory</value>
+ <description>
+ The committer factory to use when writing data to S3A filesystems.
+ If mapreduce.outputcommitter.factory.class is set, it will
+ override this property.
+ </description>
+</property>
+
</configuration>
http://git-wip-us.apache.org/repos/asf/hadoop/blob/de8b6ca5/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/lib/output/TestPathOutputCommitter.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/lib/output/TestPathOutputCommitter.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/lib/output/TestPathOutputCommitter.java
index 9cff82f..3b73934 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/lib/output/TestPathOutputCommitter.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/test/java/org/apache/hadoop/mapreduce/lib/output/TestPathOutputCommitter.java
@@ -109,14 +109,34 @@ public class TestPathOutputCommitter extends Assert {
public void abortTask(TaskAttemptContext taskContext) throws IOException {
}
+
+ @Override
+ public Path getOutputPath() {
+ return null;
+ }
}
/**
* Stub task context.
+ * The {@link #getConfiguration()} method returns the configuration supplied
+ * in the constructor; while {@link #setOutputCommitter(OutputCommitter)}
+ * sets the committer returned in {@link #getOutputCommitter()}.
+ * Otherwise, the methods are all no-ops.
*/
- public class TaskContext
+ public static class TaskContext
implements TaskInputOutputContext<String, String, String, String> {
+ private final Configuration configuration;
+
+ public TaskContext() {
+ this(new Configuration());
+ }
+
+ public TaskContext(Configuration conf) {
+ this.configuration = conf;
+ }
+
+
private OutputCommitter outputCommitter;
public void setOutputCommitter(OutputCommitter outputCommitter) {
@@ -180,7 +200,7 @@ public class TestPathOutputCommitter extends Assert {
@Override
public Configuration getConfiguration() {
- return null;
+ return configuration;
}
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org