Merge pull request #451 from agebhar1/feature/gh-447-s3-endpoint-uri

fix: add dedicated configuration for S3 endpoint override

Author: embano1

e2e/connect-config-s3.json

@@ -8,6 +8,7 @@
         "aws.eventbridge.eventbus.arn": "arn:aws:events:us-east-1:000000000000:event-bus/eventbridge-e2e",
         "aws.eventbridge.region": "us-east-1",
         "aws.eventbridge.endpoint.uri": "http://localstack:4566",
+        "aws.eventbridge.offloading.default.s3.endpoint.uri": "http://localstack:4566",
         "aws.eventbridge.offloading.default.s3.bucket": "test-bucket",
         "aws.eventbridge.offloading.default.fieldref": "$.detail.value.message",
         "key.converter": "org.apache.kafka.connect.storage.StringConverter",

src/main/java/software/amazon/event/kafkaconnector/EventBridgeSinkConfig.java

@@ -40,6 +40,8 @@ public class EventBridgeSinkConfig extends AbstractConfig {
   static final String AWS_DETAIL_TYPES_MAPPER_CLASS = "aws.eventbridge.detail.types.mapper.class";
   static final String AWS_TIME_MAPPER_CLASS = "aws.eventbridge.time.mapper.class";
   static final String AWS_EVENTBUS_RESOURCES_CONFIG = "aws.eventbridge.eventbus.resources";
+  static final String AWS_OFFLOADING_DEFAULT_S3_ENDPOINT_URI =
+      "aws.eventbridge.offloading.default.s3.endpoint.uri";
   static final String AWS_OFFLOADING_DEFAULT_S3_BUCKET =
       "aws.eventbridge.offloading.default.s3.bucket";
   static final String AWS_OFFLOADING_DEFAULT_FIELDREF =
@@ -67,6 +69,8 @@ public class EventBridgeSinkConfig extends AbstractConfig {
   private static final String AWS_PROFILE_NAME_CONFIG_DOC =
       "The profile to use from the configuration and credentials files to retrieve IAM credentials";
   public static final String AWS_DETAIL_TYPES_DEFAULT = "kafka-connect-${topic}";
+  public static final String AWS_OFFLOADING_S3_DEFAULT_ENDPOINT_URI_DOC =
+      "An optional service endpoint URI used to connect to S3.";
   public static final String AWS_OFFLOADING_S3_DEFAULT_BUCKET_DOC =
       "The S3 bucket to offload matched record value by JSON Path";
   public static final String AWS_OFFLOADING_DEFAULT_FIELDREF_DOC =
@@ -106,6 +110,7 @@ public class EventBridgeSinkConfig extends AbstractConfig {
   public final List<String> resources;
   public final int maxRetries;
   public final long retriesDelay;
+  public final String offloadingDefaultS3EndpointURI;
   public Map<String, String> detailTypeByTopic;
   public String detailType;
   public String detailTypeMapperClass;
@@ -129,6 +134,7 @@ public EventBridgeSinkConfig(final Map<?, ?> originalProps) {
     this.resources = getList(AWS_EVENTBUS_RESOURCES_CONFIG);
     this.detailTypeMapperClass = getString(AWS_DETAIL_TYPES_MAPPER_CLASS);
     this.timeMapperClass = getString(AWS_TIME_MAPPER_CLASS);
+    this.offloadingDefaultS3EndpointURI = getString(AWS_OFFLOADING_DEFAULT_S3_ENDPOINT_URI);
     this.offloadingDefaultS3Bucket = getString(AWS_OFFLOADING_DEFAULT_S3_BUCKET);
     this.offloadingDefaultFieldRef = getString(AWS_OFFLOADING_DEFAULT_FIELDREF);
 
@@ -145,7 +151,8 @@ public EventBridgeSinkConfig(final Map<?, ?> originalProps) {
         "EventBridge properties: connectorId={} eventBusArn={} eventBusRegion={} eventBusEndpointURI={} "
             + "eventBusMaxRetries={} eventBusRetriesDelay={} eventBusResources={} "
             + "eventBusEndpointID={} roleArn={} roleSessionName={} roleExternalID={} "
-            + "offloadingDefaultS3Bucket={} offloadingDefaultFieldRef={} detailTypeMapperClass={} timeMapperClass={}",
+            + "offloadingDefaultS3EndpointURI={} offloadingDefaultS3Bucket={} offloadingDefaultFieldRef={} "
+            + "detailTypeMapperClass={} timeMapperClass={}",
         connectorId,
         eventBusArn,
         region,
@@ -157,6 +164,7 @@ public EventBridgeSinkConfig(final Map<?, ?> originalProps) {
         roleArn,
         connectorId,
         externalId,
+        offloadingDefaultS3EndpointURI,
         offloadingDefaultS3Bucket,
         offloadingDefaultFieldRef,
         detailTypeMapperClass,
@@ -228,6 +236,12 @@ private static void addParams(final ConfigDef configDef) {
         AWS_DETAIL_TYPES_MAPPER_CLASS_DEFAULT,
         Importance.MEDIUM,
         AWS_DETAIL_TYPES_MAPPER_DOC);
+    configDef.define(
+        AWS_OFFLOADING_DEFAULT_S3_ENDPOINT_URI,
+        Type.STRING,
+        "",
+        Importance.MEDIUM,
+        AWS_OFFLOADING_S3_DEFAULT_ENDPOINT_URI_DOC);
     configDef.define(
         AWS_OFFLOADING_DEFAULT_S3_BUCKET,
         Type.STRING,

src/main/java/software/amazon/event/kafkaconnector/EventBridgeSinkConfigValidator.java

@@ -107,6 +107,12 @@ public static void validate(ConfigValue configValue, EnvVarGetter getenv) {
           break;
         }
 
+      case AWS_OFFLOADING_DEFAULT_S3_ENDPOINT_URI:
+        {
+          validateURI(configValue);
+          break;
+        }
+
       case AWS_OFFLOADING_DEFAULT_S3_BUCKET:
         {
           nonStrictValidateOffloadingDefaultS3Bucket(configValue);

src/main/java/software/amazon/event/kafkaconnector/EventBridgeWriter.java

@@ -70,7 +70,7 @@ public class EventBridgeWriter {
   public EventBridgeWriter(EventBridgeSinkConfig config) {
     this.config = config;
 
-    var endpointUri =
+    var ebEndpointUri =
         StringUtils.trim(this.config.endpointURI).isBlank()
             ? null
             : URI.create(this.config.endpointURI);
@@ -96,7 +96,7 @@ public EventBridgeWriter(EventBridgeSinkConfig config) {
     var client =
         EventBridgeAsyncClient.builder()
             .region(Region.of(this.config.region))
-            .endpointOverride(endpointUri)
+            .endpointOverride(ebEndpointUri)
             .httpClientBuilder(AwsCrtAsyncHttpClient.builder())
             .overrideConfiguration(clientConfig)
             .credentialsProvider(credentialsProvider)
@@ -108,11 +108,17 @@ public EventBridgeWriter(EventBridgeSinkConfig config) {
     this.batching = new DefaultEventBridgeBatching();
 
     if ((config.offloadingDefaultS3Bucket != null) && !config.offloadingDefaultS3Bucket.isEmpty()) {
+
+      var s3EndpointUri =
+          StringUtils.trim(this.config.offloadingDefaultS3EndpointURI).isBlank()
+              ? null
+              : URI.create(this.config.offloadingDefaultS3EndpointURI);
+
       var s3client =
           S3AsyncClient.builder()
               .credentialsProvider(credentialsProvider)
-              .endpointOverride(endpointUri)
-              .forcePathStyle(endpointUri != null)
+              .endpointOverride(s3EndpointUri)
+              .forcePathStyle(s3EndpointUri != null)
               .httpClientBuilder(AwsCrtAsyncHttpClient.builder())
               .overrideConfiguration(clientConfig)
               .region(Region.of(this.config.region))
@@ -121,7 +127,10 @@ public EventBridgeWriter(EventBridgeSinkConfig config) {
       var jsonPathExp = StringUtils.trim(config.offloadingDefaultFieldRef);
 
       log.info(
-          "S3 offloading is activated with bucket: {} and JSON path: {}", bucketName, jsonPathExp);
+          "S3 offloading is activated with bucket: {}, JSON path: {} and endpoint override: {}",
+          bucketName,
+          jsonPathExp,
+          s3EndpointUri == null ? "-" : s3EndpointUri);
       offloading = new S3EventBridgeEventDetailValueOffloading(s3client, bucketName, jsonPathExp);
     } else {
       log.info("S3 offloading is deactivated");