HADOOP-13786 magic commit tracker PUTs a 0-byte file to the original destination; handles apps which expect the newly written file to *exist*. It doesn't handle them trying to read it though

Author: steveloughran

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/S3AFileSystem.java

@@ -1313,6 +1313,7 @@ PutObjectRequest newPutObjectRequest(String key,
       ObjectMetadata metadata,
       InputStream inputStream) {
     Preconditions.checkNotNull(inputStream);
+    Preconditions.checkArgument(StringUtils.isNotEmpty(key), "Null/empty key");
     PutObjectRequest putObjectRequest = new PutObjectRequest(bucket, key,
         inputStream, metadata);
     setOptionalPutRequestParameters(putObjectRequest);

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/commit/PendingCommitFSIntegration.java

@@ -25,6 +25,7 @@
 
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.s3a.S3AFileSystem;
+import org.apache.hadoop.fs.s3a.commit.magic.MagicCommitTracker;
 
 import static org.apache.hadoop.fs.s3a.commit.CommitUtils.*;
 
@@ -88,7 +89,9 @@ public DefaultPutTracker getTracker(Path path, String key) {
       String pendingKey = key + CommitConstants.PENDING_SUFFIX;
       tracker = new MagicCommitTracker(path,
           owner.getBucket(),
-          destKey, pendingKey,
+          key,
+          destKey,
+          pendingKey,
           owner.createWriteOperationHelper(pendingKey));
     } else {
       // standard multipart tracking

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/commit/MagicCommitTracker.java renamed to hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/commit/magic/MagicCommitTracker.java

@@ -16,7 +16,7 @@
  * limitations under the License.
  */
 
-package org.apache.hadoop.fs.s3a.commit;
+package org.apache.hadoop.fs.s3a.commit.magic;
 
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
@@ -32,16 +32,18 @@
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.s3a.WriteOperationHelper;
+import org.apache.hadoop.fs.s3a.commit.DefaultPutTracker;
 import org.apache.hadoop.fs.s3a.commit.files.SinglePendingCommit;
 
 /**
- * Put tracker for pending commits.
+ * Put tracker for Magic commits.
  */
 @InterfaceAudience.Private
 public class MagicCommitTracker extends DefaultPutTracker {
   public static final Logger LOG = LoggerFactory.getLogger(
       MagicCommitTracker.class);
 
+  private final String originalDestKey;
   private final String pendingPartKey;
   private final Path path;
   private final WriteOperationHelper writer;
@@ -51,18 +53,21 @@ public class MagicCommitTracker extends DefaultPutTracker {
    * Pending commit tracker.
    * @param path path nominally being written to
    * @param bucket dest bucket
+   * @param originalDestKey the original key, in the magic directory.
    * @param destKey key for the destination
    * @param pendingPartKey key of the pending part
    * @param writer writer instance to use for operations
    */
   public MagicCommitTracker(Path path,
       String bucket,
+      String originalDestKey,
       String destKey,
       String pendingPartKey,
       WriteOperationHelper writer) {
     super(destKey);
     this.bucket = bucket;
     this.path = path;
+    this.originalDestKey = originalDestKey;
     this.pendingPartKey = pendingPartKey;
     this.writer = writer;
   }
@@ -114,6 +119,13 @@ public boolean aboutToComplete(String uploadId,
     PutObjectRequest put = writer.newPutRequest(
         new ByteArrayInputStream(bytes), bytes.length);
     writer.uploadObject(put);
+
+    // now put a 0-byte file with the name of the original under-magic path
+    byte[] EMPTY = new byte[0];
+    PutObjectRequest originalDestPut = writer.createPutObjectRequest(
+        originalDestKey,
+        new ByteArrayInputStream(EMPTY), 0);
+    writer.uploadObject(originalDestPut);
     return false;
   }
 

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/commit/ITestS3ACommitOperations.java

@@ -35,6 +35,7 @@
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.s3a.S3AFileSystem;
 import org.apache.hadoop.fs.s3a.commit.files.SinglePendingCommit;
+import org.apache.hadoop.fs.s3a.commit.magic.MagicCommitTracker;
 import org.apache.hadoop.fs.s3a.commit.magic.MagicS3GuardCommitter;
 import org.apache.hadoop.fs.s3a.commit.magic.MagicS3GuardCommitterFactory;
 import org.apache.hadoop.mapreduce.TaskAttemptID;