Move retries into DataSegmentPusher implementations.

docs/operations/segment-optimization.md

@@ -49,7 +49,9 @@ You may need to consider the followings to optimize your segments.
   which in turn means how well the query execution is parallelized.
   - Segment byte size: it's recommended to set 300 ~ 700MB. If this value
   doesn't match with the "number of rows per segment", please consider optimizing
-  number of rows per segment rather than this value.
+  number of rows per segment rather than this value. Note that certain deep storage
+  implementations also impose an upper limit on segment size. For example, S3 deep
+  storage imposes an upper limit of 5GB.
 
 :::info
  The above recommendation works in general, but the optimal setting can

extensions-core/hdfs-storage/src/main/java/org/apache/druid/storage/hdfs/HdfsDataSegmentPusher.java

@@ -23,11 +23,13 @@
 import com.google.common.base.Preconditions;
 import com.google.common.base.Supplier;
 import com.google.common.base.Suppliers;
+import com.google.common.base.Throwables;
 import com.google.common.collect.ImmutableMap;
 import com.google.inject.Inject;
 import org.apache.druid.common.utils.UUIDUtils;
 import org.apache.druid.guice.Hdfs;
 import org.apache.druid.java.util.common.IOE;
+import org.apache.druid.java.util.common.RetryUtils;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.common.logger.Logger;
 import org.apache.druid.segment.SegmentUtils;
@@ -113,7 +115,7 @@ public DataSegment push(final File inDir, final DataSegment segment, final boole
         uniquePrefix
     );
 
-    return pushToFilePath(inDir, segment, outIndexFilePathSuffix);
+    return pushToFilePathWithRetry(inDir, segment, outIndexFilePathSuffix);
   }
 
   @Override
@@ -125,10 +127,28 @@ public DataSegment pushToPath(File inDir, DataSegment segment, String storageDir
         storageDirSuffix.replace(':', '_'),
         segment.getShardSpec().getPartitionNum()
     );
-
-    return pushToFilePath(inDir, segment, outIndexFilePath);
+
+    return pushToFilePathWithRetry(inDir, segment, outIndexFilePath);
+  }
+
+  private DataSegment pushToFilePathWithRetry(File inDir, DataSegment segment, String outIndexFilePath)
+      throws IOException
+  {
+    // Retry any HDFS errors that occur, up to 5 times.
+    try {
+      return RetryUtils.retry(
+          () -> pushToFilePath(inDir, segment, outIndexFilePath),
+          exception -> exception instanceof Exception,
+          5
+      );
+    }
+    catch (Exception e) {
+      Throwables.throwIfInstanceOf(e, IOException.class);
+      Throwables.throwIfInstanceOf(e, RuntimeException.class);
+      throw new RuntimeException(e);
+    }
   }
-  
+
   private DataSegment pushToFilePath(File inDir, DataSegment segment, String outIndexFilePath) throws IOException
   {
     log.debug(

extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3DataSegmentPusher.java

@@ -23,6 +23,7 @@
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.inject.Inject;
+import org.apache.druid.error.DruidException;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.emitter.EmittingLogger;
 import org.apache.druid.segment.SegmentUtils;
@@ -105,6 +106,19 @@ public DataSegment pushToPath(File indexFilesDir, DataSegment inSegment, String
       );
     }
     catch (AmazonServiceException e) {
+      if (S3Utils.ERROR_ENTITY_TOO_LARGE.equals(S3Utils.getS3ErrorCode(e))) {
+        throw DruidException
+            .forPersona(DruidException.Persona.USER)
+            .ofCategory(DruidException.Category.RUNTIME_FAILURE)
+            .build(
+                e,
+                "Got error[%s] from S3 when uploading segment of size[%,d] bytes. This typically happens when segment "
+                + "size is above 5GB. Try reducing your segment size by lowering the target number of rows per "
+                + "segment.",
+                S3Utils.ERROR_ENTITY_TOO_LARGE,
+                indexSize
+            );
+      }
       throw new IOException(e);
     }
     catch (Exception e) {

extensions-core/s3-extensions/src/main/java/org/apache/druid/storage/s3/S3Utils.java

@@ -65,6 +65,11 @@ public class S3Utils
   private static final Joiner JOINER = Joiner.on("/").skipNulls();
   private static final Logger log = new Logger(S3Utils.class);
 
+  /**
+   * Error for calling putObject with an entity over 5GB in size.
+   */
+  public static final String ERROR_ENTITY_TOO_LARGE = "EntityTooLarge";
+
   public static final Predicate<Throwable> S3RETRY = new Predicate<Throwable>()
   {
     @Override
@@ -113,6 +118,18 @@ public static <T> T retryS3Operation(Task<T> f, int maxRetries) throws Exception
     return RetryUtils.retry(f, S3RETRY, maxRetries);
   }
 
+  @Nullable
+  public static String getS3ErrorCode(final Throwable e)
+  {
+    if (e == null) {
+      return null;
+    } else if (e instanceof AmazonS3Exception) {
+      return ((AmazonS3Exception) e).getErrorCode();
+    } else {
+      return getS3ErrorCode(e.getCause());
+    }
+  }
+
   static boolean isObjectInBucketIgnoringPermission(
       ServerSideEncryptingAmazonS3 s3Client,
       String bucketName,
@@ -234,11 +251,11 @@ public static ObjectMetadata getSingleObjectMetadata(ServerSideEncryptingAmazonS
   /**
    * Delete the files from S3 in a specified bucket, matching a specified prefix and filter
    *
-   * @param s3Client s3 client
-   * @param maxListingLength  maximum number of keys to fetch and delete at a time
-   * @param bucket   s3 bucket
-   * @param prefix   the file prefix
-   * @param filter   function which returns true if the prefix file found should be deleted and false otherwise.
+   * @param s3Client         s3 client
+   * @param maxListingLength maximum number of keys to fetch and delete at a time
+   * @param bucket           s3 bucket
+   * @param prefix           the file prefix
+   * @param filter           function which returns true if the prefix file found should be deleted and false otherwise.
    *
    * @throws Exception in case of errors
    */
@@ -299,7 +316,9 @@ public static void deleteBucketKeys(
       throws Exception
   {
     if (keysToDelete != null && log.isDebugEnabled()) {
-      List<String> keys = keysToDelete.stream().map(DeleteObjectsRequest.KeyVersion::getKey).collect(Collectors.toList());
+      List<String> keys = keysToDelete.stream()
+                                      .map(DeleteObjectsRequest.KeyVersion::getKey)
+                                      .collect(Collectors.toList());
       log.debug("Deleting keys from bucket: [%s], keys: [%s]", bucket, keys);
     }
     DeleteObjectsRequest deleteRequest = new DeleteObjectsRequest(bucket).withKeys(keysToDelete);

extensions-core/s3-extensions/src/test/java/org/apache/druid/storage/s3/S3DataSegmentPusherTest.java

@@ -20,61 +20,96 @@
 package org.apache.druid.storage.s3;
 
 import com.amazonaws.services.s3.model.AccessControlList;
+import com.amazonaws.services.s3.model.AmazonS3Exception;
 import com.amazonaws.services.s3.model.CanonicalGrantee;
 import com.amazonaws.services.s3.model.Grant;
 import com.amazonaws.services.s3.model.Owner;
 import com.amazonaws.services.s3.model.Permission;
 import com.amazonaws.services.s3.model.PutObjectResult;
 import com.google.common.io.Files;
+import org.apache.druid.error.DruidException;
 import org.apache.druid.java.util.common.Intervals;
 import org.apache.druid.timeline.DataSegment;
 import org.apache.druid.timeline.partition.NoneShardSpec;
 import org.easymock.EasyMock;
+import org.hamcrest.CoreMatchers;
+import org.hamcrest.MatcherAssert;
 import org.junit.Assert;
 import org.junit.Rule;
 import org.junit.Test;
+import org.junit.internal.matchers.ThrowableMessageMatcher;
 import org.junit.rules.TemporaryFolder;
 
 import java.io.File;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.function.Consumer;
 import java.util.regex.Pattern;
 
 /**
+ *
  */
 public class S3DataSegmentPusherTest
 {
-  private static class ValueContainer<T>
-  {
-    private T value;
-
-    public T getValue()
-    {
-      return value;
-    }
-
-    public void setValue(T value)
-    {
-      this.value = value;
-    }
-  }
-
   @Rule
   public final TemporaryFolder tempFolder = new TemporaryFolder();
 
   @Test
   public void testPush() throws Exception
   {
-    testPushInternal(false, "key/foo/2015-01-01T00:00:00\\.000Z_2016-01-01T00:00:00\\.000Z/0/0/index\\.zip");
+    testPushInternal(
+        false,
+        "key/foo/2015-01-01T00:00:00\\.000Z_2016-01-01T00:00:00\\.000Z/0/0/index\\.zip",
+        client ->
+            EasyMock.expect(client.putObject(EasyMock.anyObject()))
+                    .andReturn(new PutObjectResult())
+                    .once()
+    );
   }
 
   @Test
   public void testPushUseUniquePath() throws Exception
   {
-    testPushInternal(true, "key/foo/2015-01-01T00:00:00\\.000Z_2016-01-01T00:00:00\\.000Z/0/0/[A-Za-z0-9-]{36}/index\\.zip");
+    testPushInternal(
+        true,
+        "key/foo/2015-01-01T00:00:00\\.000Z_2016-01-01T00:00:00\\.000Z/0/0/[A-Za-z0-9-]{36}/index\\.zip",
+        client ->
+            EasyMock.expect(client.putObject(EasyMock.anyObject()))
+                    .andReturn(new PutObjectResult())
+                    .once()
+    );
+  }
+
+  @Test
+  public void testEntityTooLarge()
+  {
+    final DruidException exception = Assert.assertThrows(
+        DruidException.class,
+        () ->
+            testPushInternal(
+                false,
+                "key/foo/2015-01-01T00:00:00\\.000Z_2016-01-01T00:00:00\\.000Z/0/0/index\\.zip",
+                client -> {
+                  final AmazonS3Exception e = new AmazonS3Exception("whoa too many bytes");
+                  e.setErrorCode(S3Utils.ERROR_ENTITY_TOO_LARGE);
+                  EasyMock.expect(client.putObject(EasyMock.anyObject()))
+                          .andThrow(e)
+                          .once();
+                }
+            )
+    );
+
+    MatcherAssert.assertThat(
+        exception,
+        ThrowableMessageMatcher.hasMessage(CoreMatchers.startsWith("Got error[EntityTooLarge] from S3"))
+    );
   }
 
-  private void testPushInternal(boolean useUniquePath, String matcher) throws Exception
+  private void testPushInternal(
+      boolean useUniquePath,
+      String matcher,
+      Consumer<ServerSideEncryptingAmazonS3> clientDecorator
+  ) throws Exception
   {
     ServerSideEncryptingAmazonS3 s3Client = EasyMock.createStrictMock(ServerSideEncryptingAmazonS3.class);
 
@@ -83,9 +118,7 @@ private void testPushInternal(boolean useUniquePath, String matcher) throws Exce
     acl.grantAllPermissions(new Grant(new CanonicalGrantee(acl.getOwner().getId()), Permission.FullControl));
     EasyMock.expect(s3Client.getBucketAcl(EasyMock.eq("bucket"))).andReturn(acl).once();
 
-    EasyMock.expect(s3Client.putObject(EasyMock.anyObject()))
-            .andReturn(new PutObjectResult())
-            .once();
+    clientDecorator.accept(s3Client);
 
     EasyMock.replay(s3Client);
 

indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/PartialSegmentMergeTask.java

@@ -37,7 +37,6 @@
 import org.apache.druid.java.util.common.FileUtils;
 import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.java.util.common.Pair;
-import org.apache.druid.java.util.common.RetryUtils;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.common.io.Closer;
 import org.apache.druid.java.util.common.logger.Logger;
@@ -270,29 +269,24 @@ private Set<DataSegment> mergeAndPushSegments(
                                                .map(AggregatorFactory::getName)
                                                .collect(Collectors.toList());
 
-        // Retry pushing segments because uploading to deep storage might fail especially for cloud storage types
-        final DataSegment segment = RetryUtils.retry(
-            () -> segmentPusher.push(
-                mergedFileAndDimensionNames.lhs,
-                new DataSegment(
-                    getDataSource(),
-                    interval,
-                    Preconditions.checkNotNull(
-                        AbstractBatchIndexTask.findVersion(intervalToVersion, interval),
-                        "version for interval[%s]",
-                        interval
-                    ),
-                    null, // will be filled in the segmentPusher
-                    mergedFileAndDimensionNames.rhs,
-                    metricNames,
-                    createShardSpec(toolbox, interval, bucketId),
-                    null, // will be filled in the segmentPusher
-                    0     // will be filled in the segmentPusher
+        final DataSegment segment = segmentPusher.push(
+            mergedFileAndDimensionNames.lhs,
+            new DataSegment(
+                getDataSource(),
+                interval,
+                Preconditions.checkNotNull(
+                    AbstractBatchIndexTask.findVersion(intervalToVersion, interval),
+                    "version for interval[%s]",
+                    interval
                 ),
-                false
+                null, // will be filled in the segmentPusher
+                mergedFileAndDimensionNames.rhs,
+                metricNames,
+                createShardSpec(toolbox, interval, bucketId),
+                null, // will be filled in the segmentPusher
+                0     // will be filled in the segmentPusher
             ),
-            exception -> !(exception instanceof NullPointerException) && exception instanceof Exception,
-            5
+            false
         );
         long pushFinishTime = System.nanoTime();
         pushedSegments.add(segment);

processing/src/main/java/org/apache/druid/segment/loading/DataSegmentPusher.java

@@ -45,7 +45,7 @@ public interface DataSegmentPusher
   String getPathForHadoop();
 
   /**
-   * Pushes index files and segment descriptor to deep storage.
+   * Pushes index files and segment descriptor to deep storage. Expected to perform its own retries, if appropriate.
    * @param file directory containing index files
    * @param segment segment descriptor
    * @param useUniquePath if true, pushes to a unique file path. This prevents situations where task failures or replica

server/src/main/java/org/apache/druid/segment/realtime/appenderator/AppenderatorImpl.java

@@ -46,7 +46,6 @@
 import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.java.util.common.Pair;
 import org.apache.druid.java.util.common.RE;
-import org.apache.druid.java.util.common.RetryUtils;
 import org.apache.druid.java.util.common.StringUtils;
 import org.apache.druid.java.util.common.concurrent.Execs;
 import org.apache.druid.java.util.common.io.Closer;
@@ -950,19 +949,12 @@ private DataSegment mergeAndPush(
           IndexMerger.getMergedDimensionsFromQueryableIndexes(indexes, schema.getDimensionsSpec())
       );
 
-      // Retry pushing segments because uploading to deep storage might fail especially for cloud storage types
-      final DataSegment segment = RetryUtils.retry(
-          // The appenderator is currently being used for the local indexing task and the Kafka indexing task. For the
-          // Kafka indexing task, pushers must use unique file paths in deep storage in order to maintain exactly-once
-          // semantics.
-          () -> dataSegmentPusher.push(
-              mergedFile,
-              segmentToPush,
-              useUniquePath
-          ),
-          exception -> exception instanceof Exception,
-          5
-      );
+      // The appenderator is currently being used for the local indexing task and the Kafka indexing task. For the
+      // Kafka indexing task, pushers must use unique file paths in deep storage in order to maintain exactly-once
+      // semantics.
+      //
+      // dataSegmentPusher retries internally when appropriate; no need for retries here.
+      final DataSegment segment = dataSegmentPusher.push(mergedFile, segmentToPush, useUniquePath);
 
       if (!isOpenSegments()) {
         // Drop the queryable indexes behind the hydrants... they are not needed anymore and their