CXP-3043: Update kafka-connect-s3 to use AWS Java SDK 2

build.gradle

@@ -21,18 +21,20 @@ allprojects {
 
 	dependencies {
 		constraints {
-			api("com.amazonaws:aws-java-sdk-s3:1.12.351")
+			api("software.amazon.awssdk:s3:2.24.12")
 			api("org.apache.kafka:connect-api:3.5.1")
 			api("org.slf4j:slf4j-api:2.0.7")
+			api("com.fasterxml.jackson.core:jackson-databind:2.14.3")
 
 			testImplementation("junit:junit:4.13.2")
 			testImplementation("com.google.guava:guava:31.1-jre")
 			testImplementation("javax.xml.bind:jaxb-api:2.4.0-b180830.0359")
+			testImplementation("org.mockito:mockito-core:4.9.0")
 
-			testImplementation("io.debezium:debezium-testing-testcontainers:2.0.0.Final")
+			testImplementation("io.debezium:debezium-testing-testcontainers:2.5.4.Final")
 			testImplementation("net.mguenther.kafka:kafka-junit:3.3.0")
 			testImplementation("org.testcontainers:kafka:1.17.6")
-			testImplementation("org.testcontainers:localstack:1.17.6")
+			testImplementation("org.testcontainers:localstack:1.19.7")
 		}
 	}
 

common/build.gradle

@@ -3,8 +3,9 @@ description = ""
 dependencies {
     api project(':api')
 	api("org.apache.kafka:connect-api")
-	api("com.amazonaws:aws-java-sdk-s3")
+	api("software.amazon.awssdk:s3")
 	api("org.slf4j:slf4j-api")
+	api("com.fasterxml.jackson.core:jackson-databind")
 
 	testImplementation("junit:junit")
 	testImplementation("org.mockito:mockito-core:4.9.0")

common/src/main/java/com/spredfast/kafka/connect/s3/S3.java

@@ -1,15 +1,16 @@
 package com.spredfast.kafka.connect.s3;
 
-import com.amazonaws.client.builder.AwsClientBuilder.EndpointConfiguration;
-import com.amazonaws.services.s3.AmazonS3;
-import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import java.net.URI;
 import java.util.Map;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.S3ClientBuilder;
 
 public class S3 {
 
-  public static AmazonS3 s3client(Map<String, String> config) {
+  public static S3Client s3client(Map<String, String> config) {
 
-    AmazonS3ClientBuilder builder = AmazonS3ClientBuilder.standard();
+    S3ClientBuilder builder = S3Client.builder();
 
     String s3Region = config.get("s3.region");
     String s3Endpoint = config.get("s3.endpoint");
@@ -19,14 +20,15 @@ public static AmazonS3 s3client(Map<String, String> config) {
       if (s3Region == null) {
         throw new IllegalArgumentException("s3.region must be set if s3.endpoint is set");
       }
+      builder = builder.endpointOverride(URI.create(s3Endpoint));
+    }
 
-      builder = builder.withEndpointConfiguration(new EndpointConfiguration(s3Endpoint, s3Region));
-    } else if (s3Region != null) {
-      builder = builder.withRegion(s3Region);
+    if (s3Region != null) {
+      builder = builder.region(Region.of(s3Region));
     }
 
-    boolean s3PathStyle = Boolean.parseBoolean(config.get("s3.path_style"));
+    builder = builder.forcePathStyle(Boolean.parseBoolean(config.get("s3.path_style")));
 
-    return builder.withPathStyleAccessEnabled(s3PathStyle).build();
+    return builder.build();
   }
 }

sink/build.gradle

@@ -27,7 +27,7 @@ dependencies {
     implementation project(':common')
 	testImplementation("junit:junit")
 	testImplementation("org.awaitility:awaitility:4.2.0")
-	testImplementation("org.mockito:mockito-core:4.9.0")
+	testImplementation("org.mockito:mockito-core")
 	testImplementation("io.debezium:debezium-testing-testcontainers")
 	testImplementation("net.mguenther.kafka:kafka-junit")
 	testImplementation("org.testcontainers:kafka")

sink/src/main/java/com/spredfast/kafka/connect/s3/sink/S3SinkTask.java

@@ -4,7 +4,6 @@
 import static java.util.stream.Collectors.groupingBy;
 import static java.util.stream.Collectors.toList;
 
-import com.amazonaws.services.s3.AmazonS3;
 import com.spredfast.kafka.connect.s3.AlreadyBytesConverter;
 import com.spredfast.kafka.connect.s3.BlockMetadata;
 import com.spredfast.kafka.connect.s3.Configure;
@@ -33,6 +32,7 @@
 import org.apache.kafka.connect.storage.Converter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.services.s3.S3Client;
 
 public class S3SinkTask extends SinkTask {
 
@@ -102,7 +102,7 @@ public void start(Map<String, String> props) throws ConnectException {
             .filter(s -> !s.isEmpty())
             .orElseThrow(() -> new ConnectException("S3 bucket must be configured"));
     String prefix = configGet("s3.prefix").orElse("");
-    AmazonS3 s3Client = S3.s3client(config);
+    S3Client s3Client = S3.s3client(config);
 
     Layout layout = Configure.createLayout(props);
 

sink/src/main/java/com/spredfast/kafka/connect/s3/sink/S3Writer.java

@@ -1,28 +1,22 @@
 package com.spredfast.kafka.connect.s3.sink;
 
-import com.amazonaws.services.s3.AmazonS3;
-import com.amazonaws.services.s3.model.AmazonS3Exception;
-import com.amazonaws.services.s3.model.ObjectMetadata;
-import com.amazonaws.services.s3.model.PutObjectRequest;
-import com.amazonaws.services.s3.model.S3Object;
-import com.amazonaws.services.s3.transfer.TransferManager;
-import com.amazonaws.services.s3.transfer.TransferManagerBuilder;
-import com.amazonaws.services.s3.transfer.Upload;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.ObjectReader;
 import com.spredfast.kafka.connect.s3.BlockMetadata;
 import com.spredfast.kafka.connect.s3.Layout;
 import com.spredfast.kafka.connect.s3.json.ChunkDescriptor;
 import com.spredfast.kafka.connect.s3.json.ChunksIndex;
-import java.io.ByteArrayInputStream;
 import java.io.File;
 import java.io.IOException;
-import java.io.InputStreamReader;
-import java.io.Reader;
-import java.util.TimeZone;
+import java.util.function.Consumer;
 import org.apache.kafka.common.TopicPartition;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.core.sync.RequestBody;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.awssdk.services.s3.model.GetObjectRequest;
+import software.amazon.awssdk.services.s3.model.NoSuchKeyException;
+import software.amazon.awssdk.services.s3.model.PutObjectRequest;
 
 /**
  * S3Writer provides necessary operations over S3 to store files and retrieve Last commit offsets
@@ -33,38 +27,21 @@
  */
 public class S3Writer {
   private static final Logger log = LoggerFactory.getLogger(S3SinkTask.class);
-  private static final TimeZone UTC = TimeZone.getTimeZone("UTC");
   private final ObjectReader reader = new ObjectMapper().readerFor(ChunksIndex.class);
-  private String keyPrefix;
-  private String bucket;
+  private final String keyPrefix;
+  private final String bucket;
   private final Layout.Builder layoutBuilder;
-  private AmazonS3 s3Client;
-  private TransferManager tm;
+  private final S3Client s3Client;
 
   public S3Writer(
-      String bucket, String keyPrefix, Layout.Builder layoutBuilder, AmazonS3 s3Client) {
-    this(
-        bucket,
-        keyPrefix,
-        layoutBuilder,
-        s3Client,
-        TransferManagerBuilder.standard().withS3Client(s3Client).build());
-  }
-
-  public S3Writer(
-      String bucket,
-      String keyPrefix,
-      Layout.Builder layoutBuilder,
-      AmazonS3 s3Client,
-      TransferManager tm) {
-    if (keyPrefix.length() > 0 && !keyPrefix.endsWith("/")) {
+      String bucket, String keyPrefix, Layout.Builder layoutBuilder, S3Client s3Client) {
+    if (!keyPrefix.isEmpty() && !keyPrefix.endsWith("/")) {
       keyPrefix += "/";
     }
     this.keyPrefix = keyPrefix;
     this.bucket = bucket;
     this.layoutBuilder = layoutBuilder;
     this.s3Client = s3Client;
-    this.tm = tm;
   }
 
   public void putChunk(File dataFile, File indexFile, BlockMetadata metadata) throws IOException {
@@ -78,11 +55,9 @@ public void putChunk(File dataFile, File indexFile, BlockMetadata metadata) thro
     final String indexObjectKey = baseKey + ".index.json";
 
     try {
-      Upload upload = tm.upload(this.bucket, dataObjectKey, dataFile);
-      upload.waitForCompletion();
+      s3Client.putObject(putKey(dataObjectKey), dataFile.toPath());
       log.debug("uploaded {} object to s3", dataObjectKey);
-      upload = tm.upload(this.bucket, indexObjectKey, indexFile);
-      upload.waitForCompletion();
+      s3Client.putObject(putKey(indexObjectKey), indexFile.toPath());
       log.debug("uploaded {} object to s3", indexObjectKey);
     } catch (Exception e) {
       throw new IOException("Failed to upload to S3", e);
@@ -92,63 +67,52 @@ public void putChunk(File dataFile, File indexFile, BlockMetadata metadata) thro
   }
 
   public long fetchOffset(TopicPartition tp) throws IOException {
-
     // See if cursor file exists
     String indexFileKey;
+    String cursorFileKey = getCursorFileKey(tp);
 
-    try (S3Object cursorObj =
-            s3Client.getObject(this.bucket, this.getTopicPartitionLastIndexFileKey(tp));
-        InputStreamReader input = new InputStreamReader(cursorObj.getObjectContent(), "UTF-8"); ) {
-      StringBuilder sb = new StringBuilder(1024);
-      final char[] buffer = new char[1024];
-
-      for (int read = input.read(buffer, 0, buffer.length);
-          read != -1;
-          read = input.read(buffer, 0, buffer.length)) {
-        sb.append(buffer, 0, read);
-      }
-      indexFileKey = sb.toString();
-    } catch (AmazonS3Exception ase) {
-      if (ase.getStatusCode() == 404) {
-        // Topic partition has no data in S3, start from beginning
-        return 0;
-      } else {
-        throw new IOException("Failed to fetch cursor file", ase);
-      }
+    try {
+      indexFileKey = s3Client.getObjectAsBytes(getKey(cursorFileKey)).asUtf8String();
+    } catch (NoSuchKeyException e) {
+      // Topic partition has no data in S3, start from beginning
+      return 0;
     } catch (Exception e) {
       throw new IOException("Failed to fetch or read cursor file", e);
     }
 
     // Now fetch last written index file...
-    try (S3Object indexObj = s3Client.getObject(this.bucket, indexFileKey);
-        InputStreamReader isr = new InputStreamReader(indexObj.getObjectContent(), "UTF-8"); ) {
-      return getNextOffsetFromIndexFileContents(isr);
+    try {
+      byte[] indexJSON = s3Client.getObjectAsBytes(getKey(indexFileKey)).asByteArray();
+      return getNextOffsetFromIndexFileContents(indexJSON);
     } catch (Exception e) {
       throw new IOException("Failed to fetch or parse last index file", e);
     }
   }
 
-  private long getNextOffsetFromIndexFileContents(Reader indexJSON) throws IOException {
+  private long getNextOffsetFromIndexFileContents(byte[] indexJSON) throws IOException {
     ChunksIndex index = reader.readValue(indexJSON);
     ChunkDescriptor lastChunk = index.chunks.get(index.chunks.size() - 1);
     return lastChunk.first_record_offset + lastChunk.num_records;
   }
 
-  private String getTopicPartitionLastIndexFileKey(TopicPartition tp) {
+  private String getCursorFileKey(TopicPartition tp) {
     return keyPrefix + layoutBuilder.buildIndexPath(tp);
   }
 
   private void updateCursorFile(String lastIndexFileKey, TopicPartition tp) throws IOException {
+    String cursorFileKey = getCursorFileKey(tp);
     try {
-      byte[] contentAsBytes = lastIndexFileKey.getBytes("UTF-8");
-      ByteArrayInputStream contentsAsStream = new ByteArrayInputStream(contentAsBytes);
-      ObjectMetadata md = new ObjectMetadata();
-      md.setContentLength(contentAsBytes.length);
-      s3Client.putObject(
-          new PutObjectRequest(
-              this.bucket, this.getTopicPartitionLastIndexFileKey(tp), contentsAsStream, md));
+      s3Client.putObject(putKey(cursorFileKey), RequestBody.fromString(lastIndexFileKey));
     } catch (Exception ex) {
       throw new IOException("Failed to update cursor file", ex);
     }
   }
+
+  private Consumer<PutObjectRequest.Builder> putKey(String key) {
+    return builder -> builder.bucket(bucket).key(key);
+  }
+
+  private Consumer<GetObjectRequest.Builder> getKey(String key) {
+    return builder -> builder.bucket(bucket).key(key);
+  }
 }