HADOOP-18706. Improve S3ABlockOutputStream recovery (#5563)

Contributed by Chris Bevard

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

@@ -232,7 +232,9 @@ private synchronized S3ADataBlocks.DataBlock createBlockIfNeeded()
         LOG.error("Number of partitions in stream exceeds limit for S3: "
              + Constants.MAX_MULTIPART_COUNT +  " write may fail.");
       }
-      activeBlock = blockFactory.create(blockCount, this.blockSize, statistics);
+      activeBlock = blockFactory.create(
+        writeOperationHelper.getAuditSpan().getSpanId(),
+        key, blockCount, this.blockSize, statistics);
     }
     return activeBlock;
   }
@@ -728,8 +730,14 @@ public void hsync() throws IOException {
 
   /**
    * Shared processing of Syncable operation reporting/downgrade.
+   *
+   * Syncable API is not supported, so calls to hsync/hflush will throw an
+   * UnsupportedOperationException unless the stream was constructed with
+   * {@link #downgradeSyncableExceptions} set to true, in which case the stream is flushed.
+   * @throws IOException IO Problem
+   * @throws UnsupportedOperationException if downgrade syncable exceptions is set to false
    */
-  private void handleSyncableInvocation() {
+  private void handleSyncableInvocation() throws IOException {
     final UnsupportedOperationException ex
         = new UnsupportedOperationException(E_NOT_SYNCABLE);
     if (!downgradeSyncableExceptions) {
@@ -741,6 +749,7 @@ private void handleSyncableInvocation() {
         key);
     // and log at debug
     LOG.debug("Downgrading Syncable call", ex);
+    flush();
   }
 
   @Override

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

@@ -175,12 +175,14 @@ protected BlockFactory(S3AFileSystem owner) {
     /**
      * Create a block.
      *
+     * @param spanId id of the audit span
+     * @param key key of s3 object being written to
      * @param index index of block
      * @param limit limit of the block.
      * @param statistics stats to work with
      * @return a new block.
      */
-    abstract DataBlock create(long index, long limit,
+    abstract DataBlock create(String spanId, String key, long index, long limit,
         BlockOutputStreamStatistics statistics)
         throws IOException;
 
@@ -391,11 +393,11 @@ static class ArrayBlockFactory extends BlockFactory {
     }
 
     @Override
-    DataBlock create(long index, long limit,
+    DataBlock create(String spanId, String key, long index, long limit,
         BlockOutputStreamStatistics statistics)
         throws IOException {
       Preconditions.checkArgument(limit > 0,
-          "Invalid block size: %d", limit);
+          "Invalid block size: %d [%s]", limit, key);
       return new ByteArrayBlock(0, limit, statistics);
     }
 
@@ -516,11 +518,11 @@ static class ByteBufferBlockFactory extends BlockFactory {
     }
 
     @Override
-    ByteBufferBlock create(long index, long limit,
+    ByteBufferBlock create(String spanId, String key, long index, long limit,
         BlockOutputStreamStatistics statistics)
         throws IOException {
       Preconditions.checkArgument(limit > 0,
-          "Invalid block size: %d", limit);
+          "Invalid block size: %d [%s]", limit, key);
       return new ByteBufferBlock(index, limit, statistics);
     }
 
@@ -798,6 +800,8 @@ public String toString() {
    * Buffer blocks to disk.
    */
   static class DiskBlockFactory extends BlockFactory {
+    private static final String ESCAPED_FORWARD_SLASH = "EFS";
+    private static final String ESCAPED_BACKSLASH = "EBS";
 
     DiskBlockFactory(S3AFileSystem owner) {
       super(owner);
@@ -806,6 +810,8 @@ static class DiskBlockFactory extends BlockFactory {
     /**
      * Create a temp file and a {@link DiskBlock} instance to manage it.
      *
+     * @param spanId id of the audit span
+     * @param key of the s3 object being written
      * @param index block index
      * @param limit limit of the block. -1 means "no limit"
      * @param statistics statistics to update
@@ -813,17 +819,22 @@ static class DiskBlockFactory extends BlockFactory {
      * @throws IOException IO problems
      */
     @Override
-    DataBlock create(long index,
+    DataBlock create(String spanId, String key, long index,
         long limit,
         BlockOutputStreamStatistics statistics)
         throws IOException {
       Preconditions.checkArgument(limit != 0,
-          "Invalid block size: %d", limit);
+          "Invalid block size: %d [%s]", limit, key);
-      File destFile = getOwner()
+      String prefix = String.format("s3ablock-%04d-%s-%s-", index, spanId, escapeS3Key(key));
-          .createTmpFileForWrite(String.format("s3ablock-%04d-", index),
+      File destFile = getOwner().createTmpFileForWrite(prefix, limit, getOwner().getConf());
-              limit, getOwner().getConf());
       return new DiskBlock(destFile, limit, index, statistics);
     }
+
+    protected static String escapeS3Key(String key) {
+      return key
+        .replace("\\", ESCAPED_BACKSLASH)
+        .replace("/", ESCAPED_FORWARD_SLASH);
+    }
   }
 
   /**

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

@@ -217,6 +217,7 @@ public <T> T retry(String action,
    * Get the audit span this object was created with.
    * @return the audit span
    */
+  @Override
   public AuditSpan getAuditSpan() {
     return auditSpan;
   }

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

@@ -43,6 +43,7 @@
 import org.apache.hadoop.fs.PathIOException;
 import org.apache.hadoop.fs.s3a.impl.PutObjectOptions;
 import org.apache.hadoop.fs.statistics.DurationTrackerFactory;
+import org.apache.hadoop.fs.store.audit.AuditSpan;
 import org.apache.hadoop.fs.store.audit.AuditSpanSource;
 import org.apache.hadoop.util.functional.CallableRaisingIOE;
 
@@ -305,6 +306,12 @@ UploadPartResult uploadPart(UploadPartRequest request,
    */
   Configuration getConf();
 
+  /**
+   * Get the audit span this object was created with.
+   * @return the audit span
+   */
+  AuditSpan getAuditSpan();
+
   /**
    * Create a S3 Select request for the destination path.
    * This does not build the query.

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

@@ -29,9 +29,12 @@
 import org.junit.BeforeClass;
 import org.junit.Test;
 
+import java.io.File;
 import java.io.IOException;
 import java.io.InputStream;
 import java.net.URI;
+import java.util.Arrays;
+import java.util.Objects;
 
 import static org.apache.hadoop.fs.StreamCapabilities.ABORTABLE_STREAM;
 import static org.apache.hadoop.fs.s3a.Constants.*;
@@ -79,6 +82,46 @@ public void testRegularUpload() throws IOException {
     verifyUpload("regular", 1024);
   }
 
+  /**
+   * Test that the DiskBlock's local file doesn't result in error when the S3 key exceeds the max
+   * char limit of the local file system. Currently
+   * {@link java.io.File#createTempFile(String, String, File)} is being relied on to handle the
+   * truncation.
+   * @throws IOException
+   */
+  @Test
+  public void testDiskBlockCreate() throws IOException {
+    String s3Key = // 1024 char
+        "very_long_s3_key__very_long_s3_key__very_long_s3_key__very_long_s3_key__" +
+        "very_long_s3_key__very_long_s3_key__very_long_s3_key__very_long_s3_key__" +
+        "very_long_s3_key__very_long_s3_key__very_long_s3_key__very_long_s3_key__" +
+        "very_long_s3_key__very_long_s3_key__very_long_s3_key__very_long_s3_key__" +
+        "very_long_s3_key__very_long_s3_key__very_long_s3_key__very_long_s3_key__" +
+        "very_long_s3_key__very_long_s3_key__very_long_s3_key__very_long_s3_key__" +
+        "very_long_s3_key__very_long_s3_key__very_long_s3_key__very_long_s3_key__" +
+        "very_long_s3_key__very_long_s3_key__very_long_s3_key__very_long_s3_key__" +
+        "very_long_s3_key__very_long_s3_key__very_long_s3_key__very_long_s3_key__" +
+        "very_long_s3_key__very_long_s3_key__very_long_s3_key__very_long_s3_key__" +
+        "very_long_s3_key__very_long_s3_key__very_long_s3_key__very_long_s3_key__" +
+        "very_long_s3_key__very_long_s3_key__very_long_s3_key__very_long_s3_key__" +
+        "very_long_s3_key__very_long_s3_key__very_long_s3_key__very_long_s3_key__" +
+        "very_long_s3_key__very_long_s3_key__very_long_s3_key__very_long_s3_key__" +
+        "very_long_s3_key";
+    long blockSize = getFileSystem().getDefaultBlockSize();
+    try (S3ADataBlocks.BlockFactory diskBlockFactory =
+           new S3ADataBlocks.DiskBlockFactory(getFileSystem());
+         S3ADataBlocks.DataBlock dataBlock =
+             diskBlockFactory.create("spanId", s3Key, 1, blockSize, null);
+    ) {
+      String tmpDir = getConfiguration().get("hadoop.tmp.dir");
+      boolean created = Arrays.stream(
+          Objects.requireNonNull(new File(tmpDir).listFiles()))
+          .anyMatch(f -> f.getName().contains("very_long_s3_key"));
+      assertTrue(String.format("tmp file should have been created locally in %s", tmpDir), created);
+      LOG.info(dataBlock.toString()); // block file name/location can be viewed in failsafe-report
+    }
+  }
+
   @Test(expected = IOException.class)
   public void testWriteAfterStreamClose() throws Throwable {
     Path dest = path("testWriteAfterStreamClose");
@@ -136,7 +179,7 @@ private void markAndResetDatablock(S3ADataBlocks.BlockFactory factory)
         new S3AInstrumentation(new URI("s3a://example"));
     BlockOutputStreamStatistics outstats
         = instrumentation.newOutputStreamStatistics(null);
-    S3ADataBlocks.DataBlock block = factory.create(1, BLOCK_SIZE, outstats);
+    S3ADataBlocks.DataBlock block = factory.create("spanId", "object/key", 1, BLOCK_SIZE, outstats);
     block.write(dataset, 0, dataset.length);
     S3ADataBlocks.BlockUploadData uploadData = block.startUpload();
     InputStream stream = uploadData.getUploadStream();

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

@@ -51,7 +51,7 @@ public void testByteBufferIO() throws Throwable {
              new S3ADataBlocks.ByteBufferBlockFactory(null)) {
       int limit = 128;
       S3ADataBlocks.ByteBufferBlockFactory.ByteBufferBlock block
-          = factory.create(1, limit, null);
+          = factory.create("spanId", "s3\\object/key", 1, limit, null);
       assertOutstandingBuffers(factory, 1);
 
       byte[] buffer = ContractTestUtils.toAsciiByteArray("test data");

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

@@ -26,6 +26,7 @@
 import org.apache.hadoop.fs.s3a.statistics.impl.EmptyS3AStatisticsContext;
 import org.apache.hadoop.fs.s3a.test.MinimalWriteOperationHelperCallbacks;
 import org.apache.hadoop.fs.statistics.IOStatisticsContext;
+import org.apache.hadoop.fs.store.audit.AuditSpan;
 import org.apache.hadoop.util.Progressable;
 import org.junit.Before;
 import org.junit.Test;
@@ -38,7 +39,10 @@
 import static org.apache.hadoop.test.LambdaTestUtils.intercept;
 import static org.mockito.Mockito.doThrow;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
 /**
@@ -59,6 +63,9 @@ private S3ABlockOutputStream.BlockOutputStreamBuilder mockS3ABuilder() {
         mock(S3ADataBlocks.BlockFactory.class);
     long blockSize = Constants.DEFAULT_MULTIPART_SIZE;
     WriteOperationHelper oHelper = mock(WriteOperationHelper.class);
+    AuditSpan auditSpan = mock(AuditSpan.class);
+    when(auditSpan.getSpanId()).thenReturn("spanId");
+    when(oHelper.getAuditSpan()).thenReturn(auditSpan);
     PutTracker putTracker = mock(PutTracker.class);
     final S3ABlockOutputStream.BlockOutputStreamBuilder builder =
         S3ABlockOutputStream.builder()
@@ -156,6 +163,7 @@ public void testSyncableUnsupported() throws Exception {
     stream = spy(new S3ABlockOutputStream(builder));
     intercept(UnsupportedOperationException.class, () -> stream.hflush());
     intercept(UnsupportedOperationException.class, () -> stream.hsync());
+    verify(stream, never()).flush();
   }
 
   /**
@@ -169,8 +177,11 @@ public void testSyncableDowngrade() throws Exception {
     builder.withDowngradeSyncableExceptions(true);
     stream = spy(new S3ABlockOutputStream(builder));
 
+    verify(stream, never()).flush();
     stream.hflush();
+    verify(stream, times(1)).flush();
     stream.hsync();
+    verify(stream, times(2)).flush();
   }
 
 }