Skip to content

Migrate from AWS Java SDK v1 to v2 and KPL 1.x #3

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
a-bates wants to merge 2 commits into
base: master
Choose a base branch
from
Open

Migrate from AWS Java SDK v1 to v2 and KPL 1.x #3

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
babcf31
Migrate from AWS Java SDK v1 to v2 and KPL 1.x
a-bates Jun 9, 2026
7c0fab0
Pin protobuf-java to 4.29.0 for KPL 1.x compatibility
a-bates Jun 9, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
52 changes: 29 additions & 23 deletions pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,8 @@
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
<mockito.version>5.19.0</mockito.version>
<opencensus.version>0.28.3</opencensus.version>
<aws-java.version>1.12.782</aws-java.version>
<aws-sdk-v2.version>2.34.0</aws-sdk-v2.version>
<kpl.version>1.0.7</kpl.version>
<jackson.version>2.15.2</jackson.version>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
Expand Down Expand Up @@ -154,6 +155,25 @@
</profile>
</profiles>

<dependencyManagement>
<dependencies>
<dependency>
<groupId>software.amazon.awssdk</groupId>
<artifactId>bom</artifactId>
<version>${aws-sdk-v2.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
<!-- KPL 1.x generated protobuf code requires protobuf-java 4.x.
Pin here so it wins over the 3.x dragged in by google-cloud-bigquerystorage. -->
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
<version>4.29.0</version>
</dependency>
</dependencies>
</dependencyManagement>

<dependencies>
<!-- very important deps -->
<dependency>
Expand All @@ -166,11 +186,6 @@
<artifactId>c3p0</artifactId>
<version>0.9.5.5</version>
</dependency>
<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-core</artifactId>
<version>1.12.782</version>
</dependency>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>
Expand Down Expand Up @@ -276,29 +291,21 @@
<version>2.49.0</version>
</dependency>
<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-core</artifactId>
<version>1.12.782</version>
</dependency>
<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-sns</artifactId>
<version>${aws-java.version}</version>
<groupId>software.amazon.awssdk</groupId>
<artifactId>sns</artifactId>
</dependency>
<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-sqs</artifactId>
<version>${aws-java.version}</version>
<groupId>software.amazon.awssdk</groupId>
<artifactId>sqs</artifactId>
</dependency>
<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-sts</artifactId>
<version>${aws-java.version}</version>
<groupId>software.amazon.awssdk</groupId>
<artifactId>sts</artifactId>
</dependency>
<dependency>
<groupId>com.amazonaws</groupId>
<groupId>software.amazon.kinesis</groupId>
<artifactId>amazon-kinesis-producer</artifactId>
<version>0.15.12</version>
<version>${kpl.version}</version>
</dependency>
<dependency>
<groupId>javax.xml.bind</groupId>
Expand All @@ -323,7 +330,6 @@
<dependency>
<groupId>com.google.protobuf</groupId>
<artifactId>protobuf-java</artifactId>
<version>3.25.5</version>
</dependency>
<dependency>
<groupId>redis.clients</groupId>
Expand Down
11 changes: 3 additions & 8 deletions src/main/java/com/zendesk/maxwell/filtering/FilterParser.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,11 +1,10 @@
package com.zendesk.maxwell.filtering;

import com.amazonaws.util.StringInputStream;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.StreamTokenizer;
import java.io.UnsupportedEncodingException;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.regex.Pattern;
Expand All @@ -22,11 +21,7 @@ public FilterParser(String input) {
}

public List<FilterPattern> parse() throws InvalidFilterException {
try {
this.inputStream = new InputStreamReader(new StringInputStream(input));
} catch ( UnsupportedEncodingException e ) {
throw new InvalidFilterException(e.getMessage());
}
this.inputStream = new InputStreamReader(new ByteArrayInputStream(input.getBytes(StandardCharsets.UTF_8)));

this.tokenizer = new StreamTokenizer(inputStream);
try {
Expand Down
10 changes: 5 additions & 5 deletions src/main/java/com/zendesk/maxwell/producer/MaxwellKinesisProducer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,11 +18,11 @@
import com.zendesk.maxwell.replication.Position;
import com.zendesk.maxwell.row.RowMap;

import com.amazonaws.services.kinesis.producer.Attempt;
import com.amazonaws.services.kinesis.producer.KinesisProducer;
import com.amazonaws.services.kinesis.producer.KinesisProducerConfiguration;
import com.amazonaws.services.kinesis.producer.UserRecordFailedException;
import com.amazonaws.services.kinesis.producer.UserRecordResult;
import software.amazon.kinesis.producer.Attempt;
import software.amazon.kinesis.producer.KinesisProducer;
import software.amazon.kinesis.producer.KinesisProducerConfiguration;
import software.amazon.kinesis.producer.UserRecordFailedException;
import software.amazon.kinesis.producer.UserRecordResult;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down
81 changes: 40 additions & 41 deletions src/main/java/com/zendesk/maxwell/producer/MaxwellSNSProducer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,31 +1,25 @@
package com.zendesk.maxwell.producer;

import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.client.builder.AwsClientBuilder;
import com.amazonaws.handlers.AsyncHandler;
import com.amazonaws.services.sns.AmazonSNSAsync;
import com.amazonaws.services.sns.AmazonSNSAsyncClientBuilder;
import com.amazonaws.services.sns.model.MessageAttributeValue;
import com.amazonaws.services.sns.model.PublishRequest;
import com.amazonaws.services.sns.model.PublishResult;
import com.zendesk.maxwell.MaxwellContext;
import com.zendesk.maxwell.producer.partitioners.MaxwellSNSPartitioner;
import com.zendesk.maxwell.replication.Position;
import com.zendesk.maxwell.row.RowMap;
import org.apache.commons.lang3.ArrayUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.sns.SnsAsyncClient;
import software.amazon.awssdk.services.sns.model.MessageAttributeValue;
import software.amazon.awssdk.services.sns.model.PublishRequest;
import software.amazon.awssdk.services.sns.model.PublishResponse;

import java.net.URI;
import java.util.HashMap;
import java.util.Map;

public class MaxwellSNSProducer extends AbstractAsyncProducer {

private AmazonSNSAsync client;
private SnsAsyncClient client;
private String topic;
private String[] stringFelds = {"database", "table"};
private String[] numberFields = {"ts", "xid"};
private MaxwellSNSPartitioner partitioner;

public MaxwellSNSProducer(MaxwellContext context, String topic, String serviceEndpoint, String signingRegion) {
Expand All @@ -35,64 +29,69 @@ public MaxwellSNSProducer(MaxwellContext context, String topic, String serviceEn
// Only configure custom endpoint if both serviceEndpoint and signingRegion are provided
if (serviceEndpoint != null && !serviceEndpoint.trim().isEmpty() &&
signingRegion != null && !signingRegion.trim().isEmpty()) {
this.client = AmazonSNSAsyncClientBuilder.standard()
.withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(serviceEndpoint, signingRegion))
this.client = SnsAsyncClient.builder()
.endpointOverride(URI.create(serviceEndpoint))
.region(Region.of(signingRegion))
.build();
} else {
// Use default client configuration when endpoint parameters are not provided
this.client = AmazonSNSAsyncClientBuilder.defaultClient();
this.client = SnsAsyncClient.create();
}
String partitionKey = context.getConfig().producerPartitionKey;
String partitionColumns = context.getConfig().producerPartitionColumns;
String partitionFallback = context.getConfig().producerPartitionFallback;
this.partitioner = new MaxwellSNSPartitioner(partitionKey, partitionColumns, partitionFallback);
}

public void setClient(AmazonSNSAsync client) {
public void setClient(SnsAsyncClient client) {
this.client = client;
}

@Override
public void sendAsync(RowMap r, CallbackCompleter cc) throws Exception {
String value = r.toJSON(outputConfig);
// Publish a message to an Amazon SNS topic.
final PublishRequest publishRequest = new PublishRequest(topic, value);
Map<String, MessageAttributeValue> messageAttributes = new HashMap<>();

Map<String, MessageAttributeValue> messageAttributes = new HashMap<>();
final String configuredAttributes = context.getConfig().snsAttrs;
if (configuredAttributes != null) {
for (String element: configuredAttributes.split(",")) {
switch (element) {
case "database":
messageAttributes.put(
"database",
new MessageAttributeValue().withDataType("String").withStringValue(r.getDatabase())
MessageAttributeValue.builder().dataType("String").stringValue(r.getDatabase()).build()
);
break;
case "table":
messageAttributes.put(
"table",
new MessageAttributeValue().withDataType("String").withStringValue(r.getTable())
MessageAttributeValue.builder().dataType("String").stringValue(r.getTable()).build()
);
break;
}
}
}

if ( topic.endsWith(".fifo")) {
PublishRequest.Builder builder = PublishRequest.builder()
.topicArn(topic)
.message(value)
.messageAttributes(messageAttributes);

if (topic.endsWith(".fifo")) {
String key = this.partitioner.getSNSKey(r);
publishRequest.setMessageGroupId(key);
builder.messageGroupId(key);
}
publishRequest.setMessageAttributes(messageAttributes);

PublishRequest publishRequest = builder.build();

SNSCallback callback = new SNSCallback(cc, r.getNextPosition(), value,
r.getDatabase(), r.getTable(), r.getRowIdentity().toConcatString(), r.getApproximateSize(), context);
client.publishAsync(publishRequest, callback);
client.publish(publishRequest).whenComplete(callback);
}

}

class SNSCallback implements AsyncHandler<PublishRequest, PublishResult> {
class SNSCallback implements java.util.function.BiConsumer<PublishResponse, Throwable> {
public static final Logger logger = LoggerFactory.getLogger(SNSCallback.class);

private final AbstractAsyncProducer.CallbackCompleter cc;
Expand All @@ -118,23 +117,23 @@ public SNSCallback(AbstractAsyncProducer.CallbackCompleter cc, Position position
}

@Override
public void onError(Exception t) {
logger.error(t.getClass().getSimpleName() + " @ " + position + " -- ");
logger.error("Database:" + database + ", Table:" + table + ", PK:" + pk + ", Approx Size:" + Long.toString(approximateRowSize));
logger.error(t.getLocalizedMessage());
logger.error("Exception during put", t);

if (!context.getConfig().ignoreProducerError) {
context.terminate(new RuntimeException(t));
} else {
cc.markCompleted();
public void accept(PublishResponse response, Throwable t) {
if (t != null) {
logger.error(t.getClass().getSimpleName() + " @ " + position + " -- ");
logger.error("Database:" + database + ", Table:" + table + ", PK:" + pk + ", Approx Size:" + Long.toString(approximateRowSize));
logger.error(t.getLocalizedMessage());
logger.error("Exception during put", t);

if (!context.getConfig().ignoreProducerError) {
context.terminate(new RuntimeException(t));
} else {
cc.markCompleted();
}
return;
}
};

@Override
public void onSuccess(PublishRequest request, PublishResult result) {
if (logger.isDebugEnabled()) {
logger.debug("-> MessageId: {}", result.getMessageId());
logger.debug("-> MessageId: {}", response.messageId());
}
cc.markCompleted();
}
Expand Down
Loading
Loading