add endpoint override support

debezium · Dec 4, 2023 · ea03bde · ea03bde
1 parent e71a359
commit ea03bde
Showing 1 changed file with 5 additions and 0 deletions.
diff --git a/debezium-server-sqs/src/main/java/io/debezium/server/sqs/SqsChangeConsumer.java b/debezium-server-sqs/src/main/java/io/debezium/server/sqs/SqsChangeConsumer.java
@@ -5,6 +5,7 @@
  */
 package io.debezium.server.sqs;
 
+import java.net.URI;
 import java.time.Duration;
 import java.util.List;
 import java.util.Optional;
@@ -51,10 +52,12 @@ public class SqsChangeConsumer extends BaseChangeConsumer implements ChangeConsu
 
     protected static final String PROP_PREFIX = "debezium.sink.sqs.";
     protected static final String PROP_REGION_NAME = PROP_PREFIX + "region";
+    private static final String PROP_ENDPOINT_NAME = PROP_PREFIX + "endpoint";
     private static final String PROP_QUEUE_URL = PROP_PREFIX + "queue.url";
     private static final String PROP_CREDENTIALS_PROFILE = PROP_PREFIX + "credentials.profile";
 
     private String region;
+    private Optional<String> endpointOverride;
     private String queueUrl;
 
     private Boolean isFifoQueue;
@@ -84,10 +87,12 @@ void connect() {
 
         final Config config = ConfigProvider.getConfig();
         region = config.getValue(PROP_REGION_NAME, String.class);
+        endpointOverride = config.getOptionalValue(PROP_ENDPOINT_NAME, String.class);
         queueUrl = config.getValue(PROP_QUEUE_URL, String.class);
         isFifoQueue = queueUrl.endsWith(".fifo");
         final SqsClientBuilder builder = SqsClient.builder()
                 .region(Region.of(region));
+        endpointOverride.ifPresent(endpoint -> builder.endpointOverride(URI.create(endpoint)));
         credentialsProfile.ifPresent(profile -> builder.credentialsProvider(ProfileCredentialsProvider.create(profile)));
 
         client = builder.build();