More changes

Author: jtuglu1

cloud/aws-common/src/main/java/org/apache/druid/common/aws/FileSessionCredentialsProvider.java

@@ -32,6 +32,16 @@
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 
+/**
+ * Loads AWS session credentials from a properties file and refreshes them periodically.
+ *
+ * <p>The credentials file must contain {@code sessionToken}, {@code accessKey}, and {@code secretKey} properties.
+ * Credentials are loaded immediately on construction and refreshed every hour via a background thread.
+ *
+ * <p>Note: In AWS SDK v1, {@code AWSCredentialsProvider} had a {@code refresh()} method as part of the interface.
+ * In SDK v2, {@code AwsCredentialsProvider} has no such method, so refresh is handled internally via a
+ * scheduled executor.
+ */
 public class FileSessionCredentialsProvider implements AwsCredentialsProvider
 {
   private final ScheduledExecutorService scheduler =

cloud/aws-common/src/test/java/org/apache/druid/common/aws/AWSClientUtilTest.java

@@ -21,6 +21,7 @@
 
 import org.junit.Assert;
 import org.junit.Test;
+import software.amazon.awssdk.awscore.exception.AwsErrorDetails;
 import software.amazon.awssdk.awscore.exception.AwsServiceException;
 import software.amazon.awssdk.core.exception.SdkClientException;
 
@@ -39,7 +40,7 @@ public void testRecoverableException_RequestTimeout()
   {
     AwsServiceException ex = AwsServiceException.builder()
         .message("RequestTimeout")
-        .awsErrorDetails(software.amazon.awssdk.awscore.exception.AwsErrorDetails.builder()
+        .awsErrorDetails(AwsErrorDetails.builder()
             .errorCode("RequestTimeout")
             .build())
         .build();
@@ -81,7 +82,7 @@ public void testRecoverableException_ProvisionedThroughputExceededException()
   {
     AwsServiceException ex = AwsServiceException.builder()
         .message("ProvisionedThroughputExceededException")
-        .awsErrorDetails(software.amazon.awssdk.awscore.exception.AwsErrorDetails.builder()
+        .awsErrorDetails(AwsErrorDetails.builder()
             .errorCode("ProvisionedThroughputExceededException")
             .build())
         .build();
@@ -93,7 +94,7 @@ public void testRecoverableException_ClockSkewedError()
   {
     AwsServiceException ex = AwsServiceException.builder()
         .message("RequestExpired")
-        .awsErrorDetails(software.amazon.awssdk.awscore.exception.AwsErrorDetails.builder()
+        .awsErrorDetails(AwsErrorDetails.builder()
             .errorCode("RequestExpired")
             .build())
         .build();

extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplier.java

@@ -521,7 +521,8 @@ public static KinesisClient getAmazonKinesisClient(
    * Parse region from a Kinesis endpoint URL.
    * Expected format: https://kinesis.{region}.amazonaws.com
    */
-  private static Region parseRegionFromEndpoint(String endpoint)
+  @VisibleForTesting
+  static Region parseRegionFromEndpoint(String endpoint)
   {
     if (endpoint == null) {
       return null;

extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisRecordSupplierTest.java

@@ -37,6 +37,7 @@
 import org.junit.Test;
 import software.amazon.awssdk.core.SdkBytes;
 import software.amazon.awssdk.core.exception.SdkClientException;
+import software.amazon.awssdk.regions.Region;
 import software.amazon.awssdk.services.kinesis.KinesisClient;
 import software.amazon.awssdk.services.kinesis.model.DescribeStreamRequest;
 import software.amazon.awssdk.services.kinesis.model.DescribeStreamResponse;
@@ -837,4 +838,51 @@ public void testIsOffsetAvailable()
 
     target.close();
   }
+
+  @Test
+  public void testParseRegionFromEndpoint_standardEndpoint()
+  {
+    Assert.assertEquals(
+        Region.US_EAST_1,
+        KinesisRecordSupplier.parseRegionFromEndpoint("https://kinesis.us-east-1.amazonaws.com")
+    );
+  }
+
+  @Test
+  public void testParseRegionFromEndpoint_withoutScheme()
+  {
+    Assert.assertEquals(
+        Region.EU_WEST_1,
+        KinesisRecordSupplier.parseRegionFromEndpoint("kinesis.eu-west-1.amazonaws.com")
+    );
+  }
+
+  @Test
+  public void testParseRegionFromEndpoint_cnRegion()
+  {
+    Assert.assertEquals(
+        Region.of("cn-north-1"),
+        KinesisRecordSupplier.parseRegionFromEndpoint("https://kinesis.cn-north-1.amazonaws.com")
+    );
+  }
+
+  @Test
+  public void testParseRegionFromEndpoint_null()
+  {
+    Assert.assertNull(KinesisRecordSupplier.parseRegionFromEndpoint(null));
+  }
+
+  @Test
+  public void testParseRegionFromEndpoint_nonAwsEndpoint()
+  {
+    Assert.assertNull(KinesisRecordSupplier.parseRegionFromEndpoint("https://localhost:4566"));
+  }
+
+  @Test
+  public void testParseRegionFromEndpoint_noKinesisPrefix()
+  {
+    Assert.assertNull(
+        KinesisRecordSupplier.parseRegionFromEndpoint("https://custom.us-east-1.amazonaws.com")
+    );
+  }
 }

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

@@ -35,7 +35,7 @@
  * Iterator class used by {@link S3Utils#objectSummaryIterator}.
  *
  * As required by the specification of that method, this iterator is computed incrementally in batches of
- * {@code maxListLength}. The first call is made at the same time the iterator is constructed.
+ * {@code maxListLength}. The first S3 call is made lazily on the first call to {@link #hasNext()} or {@link #next()}.
  */
 public class ObjectSummaryIterator implements Iterator<S3Object>
 {
@@ -50,7 +50,7 @@ public class ObjectSummaryIterator implements Iterator<S3Object>
   private Iterator<S3Object> objectSummaryIterator;
   private S3Object currentObjectSummary;
   private int maxRetries; // this is made available for testing mostly
-
+  private boolean initialized;
 
   ObjectSummaryIterator(
       final ServerSideEncryptingAmazonS3 s3Client,
@@ -62,9 +62,6 @@ public class ObjectSummaryIterator implements Iterator<S3Object>
     this.prefixesIterator = prefixes.iterator();
     this.maxListingLength = maxListingLength;
     maxRetries = RetryUtils.DEFAULT_MAX_TRIES;
-
-    constructorPostProcessing();
-
   }
 
   @VisibleForTesting
@@ -79,28 +76,29 @@ public class ObjectSummaryIterator implements Iterator<S3Object>
     this.prefixesIterator = prefixes.iterator();
     this.maxListingLength = maxListingLength;
     this.maxRetries = maxRetries;
-
-    constructorPostProcessing();
-
   }
 
-  // helper to factor out stuff that happens in constructor after members are set
-  private void constructorPostProcessing()
+  private void initialize()
   {
-    prepareNextRequest();
-    fetchNextBatch();
-    advanceObjectSummary();
+    if (!initialized) {
+      initialized = true;
+      prepareNextRequest();
+      fetchNextBatch();
+      advanceObjectSummary();
+    }
   }
 
   @Override
   public boolean hasNext()
   {
+    initialize();
     return currentObjectSummary != null;
   }
 
   @Override
   public S3Object next()
   {
+    initialize();
     if (currentObjectSummary == null) {
       throw new NoSuchElementException();
     }

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

@@ -46,6 +46,7 @@ public class ObjectSummaryWithBucketIterator implements Iterator<S3Utils.S3Objec
   private ListObjectsV2Response result;
   private Iterator<S3Object> objectSummaryIterator;
   private S3Utils.S3ObjectWithBucket currentObjectSummary;
+  private boolean initialized;
 
   ObjectSummaryWithBucketIterator(
       final ServerSideEncryptingAmazonS3 s3Client,
@@ -58,21 +59,29 @@ public class ObjectSummaryWithBucketIterator implements Iterator<S3Utils.S3Objec
     this.prefixesIterator = prefixes.iterator();
     this.maxListingLength = maxListingLength;
     this.maxRetries = maxRetries;
-
-    prepareNextRequest();
-    fetchNextBatch();
-    advanceObjectSummary();
   }
 
   @Override
   public boolean hasNext()
   {
+    initialize();
     return currentObjectSummary != null;
   }
 
+  private void initialize()
+  {
+    if (!initialized) {
+      initialized = true;
+      prepareNextRequest();
+      fetchNextBatch();
+      advanceObjectSummary();
+    }
+  }
+
   @Override
   public S3Utils.S3ObjectWithBucket next()
   {
+    initialize();
     if (currentObjectSummary == null) {
       throw new NoSuchElementException();
     }
@@ -148,6 +157,13 @@ private void advanceObjectSummary()
     currentObjectSummary = null;
   }
 
+  /**
+   * Checks if a given object is a directory placeholder and should be ignored.
+   *
+   * Adapted from org.jets3t.service.model.StorageObject.isDirectoryPlaceholder(). Does not include the check for
+   * legacy JetS3t directory placeholder objects, since it is based on content-type, which isn't available in an
+   * S3Object.
+   */
   private static boolean isDirectoryPlaceholder(final S3Object objectSummary)
   {
     // Recognize "standard" directory place-holder indications used by Amazon's AWS Console and Panic's Transmit.

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

@@ -29,6 +29,7 @@
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
+import software.amazon.awssdk.awscore.exception.AwsErrorDetails;
 import software.amazon.awssdk.core.ResponseInputStream;
 import software.amazon.awssdk.http.AbortableInputStream;
 import software.amazon.awssdk.services.s3.model.GetObjectResponse;
@@ -145,7 +146,7 @@ public void testGZUncompressOn4xxError() throws IOException
 
     S3Exception exception = (S3Exception) S3Exception.builder()
         .message("S3DataSegmentPullerTest")
-        .awsErrorDetails(software.amazon.awssdk.awscore.exception.AwsErrorDetails.builder()
+        .awsErrorDetails(AwsErrorDetails.builder()
             .errorCode("NoSuchKey")
             .errorMessage("S3DataSegmentPullerTest")
             .build())
@@ -194,7 +195,7 @@ public void testGZUncompressOn5xxError() throws IOException, SegmentLoadingExcep
 
     S3Exception exception = (S3Exception) S3Exception.builder()
         .message("S3DataSegmentPullerTest")
-        .awsErrorDetails(software.amazon.awssdk.awscore.exception.AwsErrorDetails.builder()
+        .awsErrorDetails(AwsErrorDetails.builder()
             .errorCode("Slow Down")
             .errorMessage("S3DataSegmentPullerTest")
             .build())