Support s3 path style access

README.md

@@ -115,7 +115,7 @@ for backwards compatibility but leads to increased memory usage. It is recommend
    * Default: ""
    * Importance: low
 
-  
+
  ``large.message.s3.role.external.id``
    AWS STS role external ID used when retrieving session credentials under an assumed role. Leave empty if AWS Basic provider or AWS credential provider chain should be used.
 
@@ -149,7 +149,13 @@ Endpoint to use for connection to Amazon S3. Leave empty if default S3 endpoint
 
   * Type: string
   * Default: ""
-  * Importance: low
+
+ ``large.message.s3.path.style.access``
+  Enable path-style access for S3 client.
+
+  * Type: boolean
+  * Default: false
+  * Importance: low * Importance: low
 
 ``large.message.abs.connection.string``
   Azure connection string for connection to blob storage. Leave empty if Azure credential provider chain should be used.
@@ -161,7 +167,7 @@ Endpoint to use for connection to Amazon S3. Leave empty if default S3 endpoint
 ``large.message.gs.key.path``
   Google service account key JSON path. Leave empty If the environment variable GOOGLE_APPLICATION_CREDENTIALS is set
   or if you want to use the default service account provided by your computer engine. For more information about
-  authenticating as a service account please 
+  authenticating as a service account please
   refer to the [main documentation](https://cloud.google.com/docs/authentication/production).
 
   * Type: string

large-message-core/src/main/java/com/bakdata/kafka/AbstractLargeMessageConfig.java

@@ -141,6 +141,9 @@ public class AbstractLargeMessageConfig extends AbstractConfig {
     public static final String S3_ENDPOINT_DOC =
             "Endpoint to use for connection to Amazon S3. Leave empty if default S3 endpoint should be used.";
     public static final String S3_ENDPOINT_DEFAULT = "";
+    public static final String S3_ENABLE_PATH_STYLE_ACCESS_CONFIG = S3_PREFIX + "path.style.access";
+    public static final String S3_ENABLE_PATH_STYLE_ACCESS_DOC = "Enable path-style access for S3 client.";
+    public static final boolean S3_ENABLE_PATH_STYLE_ACCESS_DEFAULT = false;
     public static final String S3_REGION_DEFAULT = "";
     public static final String S3_ACCESS_KEY_DOC = "AWS access key to use for connecting to S3. Leave empty if AWS"
             + " credential provider chain or STS Assume Role provider should be used.";
@@ -207,6 +210,8 @@ protected static ConfigDef baseConfigDef() {
                         COMPRESSION_TYPE_DOC)
                 // Amazon S3
                 .define(S3_ENDPOINT_CONFIG, Type.STRING, S3_ENDPOINT_DEFAULT, Importance.LOW, S3_ENDPOINT_DOC)
+                .define(S3_ENABLE_PATH_STYLE_ACCESS_CONFIG, Type.BOOLEAN, S3_ENABLE_PATH_STYLE_ACCESS_DEFAULT,
+                        Importance.LOW, S3_ENABLE_PATH_STYLE_ACCESS_DOC)
                 .define(S3_REGION_CONFIG, Type.STRING, S3_REGION_DEFAULT, Importance.LOW, S3_REGION_DOC)
                 .define(S3_ACCESS_KEY_CONFIG, Type.PASSWORD, S3_ACCESS_KEY_DEFAULT, Importance.LOW, S3_ACCESS_KEY_DOC)
                 .define(S3_SECRET_KEY_CONFIG, Type.PASSWORD, S3_SECRET_KEY_DEFAULT, Importance.LOW, S3_SECRET_KEY_DOC)
@@ -296,6 +301,9 @@ private BlobStorageClient createAmazonS3Client() {
         this.getAmazonEndpointOverride().ifPresent(clientBuilder::endpointOverride);
         this.getAmazonRegion().ifPresent(clientBuilder::region);
         this.getAmazonCredentialsProvider().ifPresent(clientBuilder::credentialsProvider);
+        if (this.enableAmazonS3PathStyleAccess()) {
+            clientBuilder.forcePathStyle(true);
+        }
         return new AmazonS3Client(clientBuilder.build());
     }
 
@@ -304,6 +312,10 @@ private Optional<URI> getAmazonEndpointOverride() {
         return isEmpty(endpoint) ? Optional.empty() : Optional.of(URI.create(endpoint));
     }
 
+    private boolean enableAmazonS3PathStyleAccess() {
+        return this.getBoolean(S3_ENABLE_PATH_STYLE_ACCESS_CONFIG);
+    }
+
     private Optional<Region> getAmazonRegion() {
         final String region = this.getString(S3_REGION_CONFIG);
         return isEmpty(region) ? Optional.empty() : Optional.of(Region.of(region));

large-message-core/src/test/java/com/bakdata/kafka/AmazonS3LargeMessageClientRoundtripTest.java

@@ -30,6 +30,8 @@
 import io.confluent.common.config.ConfigDef;
 import java.util.Map;
 import java.util.stream.Stream;
+import lombok.Builder;
+import lombok.Value;
 import org.apache.kafka.common.header.Header;
 import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.header.internals.RecordHeaders;
@@ -52,18 +54,25 @@ private static byte[] serialize(final String s) {
 
     private static Stream<Arguments> provideParameters() {
         return Stream.of(true, false)
-                .flatMap(isKey -> Stream.of("none", "gzip", "snappy", "lz4", "zstd").map(c -> Arguments.of(isKey, c)));
+                .flatMap(isKey -> Stream.of(true, false)
+                        .map(enabledPathAccess -> RoundtripArgument.builder()
+                                .isKey(isKey)
+                                .isPathStyleAccess(enabledPathAccess)))
+                .flatMap(builder -> Stream.of("none", "gzip", "snappy", "lz4", "zstd")
+                        .map(c -> builder.compressionType(c).build())
+                        .map(Arguments::of));
     }
 
     @ParameterizedTest
     @MethodSource("provideParameters")
-    void shouldRoundtrip(final boolean isKey, final String compressionType) {
+    void shouldRoundtrip(final RoundtripArgument argument) {
         final String bucket = "bucket";
         final String basePath = "s3://" + bucket + "/base/";
         final Map<String, Object> properties = ImmutableMap.<String, Object>builder()
                 .put(AbstractLargeMessageConfig.MAX_BYTE_SIZE_CONFIG, 0)
                 .put(AbstractLargeMessageConfig.BASE_PATH_CONFIG, basePath)
-                .put(AbstractLargeMessageConfig.COMPRESSION_TYPE_CONFIG, compressionType)
+                .put(AbstractLargeMessageConfig.S3_ENABLE_PATH_STYLE_ACCESS_CONFIG, argument.isPathStyleAccess())
+                .put(AbstractLargeMessageConfig.COMPRESSION_TYPE_CONFIG, argument.getCompressionType())
                 .build();
         final S3Client s3 = this.getS3Client();
         s3.createBucket(CreateBucketRequest.builder().bucket(bucket).build());
@@ -73,16 +82,16 @@ void shouldRoundtrip(final boolean isKey, final String compressionType) {
 
         final Headers headers = new RecordHeaders();
         final byte[] obj = serialize("big value");
-        final byte[] data = storer.storeBytes(TOPIC, obj, isKey, headers);
+        final byte[] data = storer.storeBytes(TOPIC, obj, argument.isKey(), headers);
 
         final Iterable<Header> compressionHeaders = headers.headers(CompressionType.HEADER_NAME);
-        if ("none".equals(compressionType)) {
+        if ("none".equals(argument.getCompressionType())) {
             assertThat(compressionHeaders).isEmpty();
         } else {
             assertThat(compressionHeaders).isNotEmpty();
         }
 
-        final byte[] result = retriever.retrieveBytes(data, headers, isKey);
+        final byte[] result = retriever.retrieveBytes(data, headers, argument.isKey());
         assertThat(result).isEqualTo(obj);
     }
 
@@ -105,4 +114,12 @@ private LargeMessageRetrievingClient createRetriever() {
         final AbstractLargeMessageConfig config = new AbstractLargeMessageConfig(configDef, properties);
         return config.getRetriever();
     }
+
+    @Builder
+    @Value
+    static class RoundtripArgument {
+        boolean isKey;
+        boolean isPathStyleAccess;
+        String compressionType;
+    }
 }