diff --git a/pom.xml b/pom.xml index d0a4e9f9a..e4c0f29bf 100644 --- a/pom.xml +++ b/pom.xml @@ -43,7 +43,8 @@ UTF-8 5.19.0 0.28.3 - 1.12.782 + 2.34.0 + 1.0.7 2.15.2 1.8 1.8 @@ -154,6 +155,25 @@ + + + + software.amazon.awssdk + bom + ${aws-sdk-v2.version} + pom + import + + + + com.google.protobuf + protobuf-java + 4.29.0 + + + + @@ -166,11 +186,6 @@ c3p0 0.9.5.5 - - com.amazonaws - aws-java-sdk-core - 1.12.782 - commons-io commons-io @@ -276,29 +291,21 @@ 2.49.0 - com.amazonaws - aws-java-sdk-core - 1.12.782 - - - com.amazonaws - aws-java-sdk-sns - ${aws-java.version} + software.amazon.awssdk + sns - com.amazonaws - aws-java-sdk-sqs - ${aws-java.version} + software.amazon.awssdk + sqs - com.amazonaws - aws-java-sdk-sts - ${aws-java.version} + software.amazon.awssdk + sts - com.amazonaws + software.amazon.kinesis amazon-kinesis-producer - 0.15.12 + ${kpl.version} javax.xml.bind @@ -323,7 +330,6 @@ com.google.protobuf protobuf-java - 3.25.5 redis.clients diff --git a/src/main/java/com/zendesk/maxwell/filtering/FilterParser.java b/src/main/java/com/zendesk/maxwell/filtering/FilterParser.java index 821126078..84ce3ff78 100644 --- a/src/main/java/com/zendesk/maxwell/filtering/FilterParser.java +++ b/src/main/java/com/zendesk/maxwell/filtering/FilterParser.java @@ -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; @@ -22,11 +21,7 @@ public FilterParser(String input) { } public List 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 { diff --git a/src/main/java/com/zendesk/maxwell/producer/MaxwellKinesisProducer.java b/src/main/java/com/zendesk/maxwell/producer/MaxwellKinesisProducer.java index 07489cf17..447838b61 100644 --- a/src/main/java/com/zendesk/maxwell/producer/MaxwellKinesisProducer.java +++ b/src/main/java/com/zendesk/maxwell/producer/MaxwellKinesisProducer.java @@ -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; diff --git a/src/main/java/com/zendesk/maxwell/producer/MaxwellSNSProducer.java b/src/main/java/com/zendesk/maxwell/producer/MaxwellSNSProducer.java index 9aa4688b3..a839619ad 100644 --- a/src/main/java/com/zendesk/maxwell/producer/MaxwellSNSProducer.java +++ b/src/main/java/com/zendesk/maxwell/producer/MaxwellSNSProducer.java @@ -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) { @@ -35,12 +29,13 @@ 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; @@ -48,17 +43,15 @@ public MaxwellSNSProducer(MaxwellContext context, String topic, String serviceEn 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 messageAttributes = new HashMap<>(); + Map messageAttributes = new HashMap<>(); final String configuredAttributes = context.getConfig().snsAttrs; if (configuredAttributes != null) { for (String element: configuredAttributes.split(",")) { @@ -66,33 +59,39 @@ public void sendAsync(RowMap r, CallbackCompleter cc) throws Exception { 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 { +class SNSCallback implements java.util.function.BiConsumer { public static final Logger logger = LoggerFactory.getLogger(SNSCallback.class); private final AbstractAsyncProducer.CallbackCompleter cc; @@ -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(); } diff --git a/src/main/java/com/zendesk/maxwell/producer/MaxwellSQSProducer.java b/src/main/java/com/zendesk/maxwell/producer/MaxwellSQSProducer.java index 027afe5d6..838fadc5a 100644 --- a/src/main/java/com/zendesk/maxwell/producer/MaxwellSQSProducer.java +++ b/src/main/java/com/zendesk/maxwell/producer/MaxwellSQSProducer.java @@ -1,30 +1,30 @@ package com.zendesk.maxwell.producer; -import com.amazonaws.client.builder.AwsClientBuilder; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import com.amazonaws.handlers.AsyncHandler; -import com.amazonaws.services.sqs.AmazonSQSAsync; -import com.amazonaws.services.sqs.AmazonSQSAsyncClientBuilder; -import com.amazonaws.services.sqs.model.SendMessageRequest; -import com.amazonaws.services.sqs.model.SendMessageResult; import com.zendesk.maxwell.MaxwellContext; import com.zendesk.maxwell.producer.partitioners.MaxwellSQSPartitioner; import com.zendesk.maxwell.replication.Position; import com.zendesk.maxwell.row.RowMap; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import software.amazon.awssdk.regions.Region; +import software.amazon.awssdk.services.sqs.SqsAsyncClient; +import software.amazon.awssdk.services.sqs.model.SendMessageRequest; +import software.amazon.awssdk.services.sqs.model.SendMessageResponse; + +import java.net.URI; public class MaxwellSQSProducer extends AbstractAsyncProducer { - private AmazonSQSAsync client; + private SqsAsyncClient client; private String queueUri; private MaxwellSQSPartitioner partitioner; public MaxwellSQSProducer(MaxwellContext context, String queueUri, String serviceEndpoint, String signingRegion) { super(context); this.queueUri = queueUri; - this.client = AmazonSQSAsyncClientBuilder.standard() - .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(serviceEndpoint, signingRegion)) + this.client = SqsAsyncClient.builder() + .endpointOverride(URI.create(serviceEndpoint)) + .region(Region.of(signingRegion)) .build(); String partitionKey = context.getConfig().producerPartitionKey; String partitionColumns = context.getConfig().producerPartitionColumns; @@ -35,18 +35,24 @@ public MaxwellSQSProducer(MaxwellContext context, String queueUri, String servic @Override public void sendAsync(RowMap r, CallbackCompleter cc) throws Exception { String value = r.toJSON(outputConfig); - SendMessageRequest messageRequest = new SendMessageRequest(queueUri, value); - if ( queueUri.endsWith(".fifo")) { + + SendMessageRequest.Builder builder = SendMessageRequest.builder() + .queueUrl(queueUri) + .messageBody(value); + + if (queueUri.endsWith(".fifo")) { String key = this.partitioner.getSQSKey(r); - messageRequest.setMessageGroupId(key); + builder.messageGroupId(key); } + + SendMessageRequest messageRequest = builder.build(); SQSCallback callback = new SQSCallback(cc, r.getNextPosition(), value, context); - client.sendMessageAsync(messageRequest, callback); + client.sendMessage(messageRequest).whenComplete(callback); } } -class SQSCallback implements AsyncHandler { +class SQSCallback implements java.util.function.BiConsumer { public static final Logger logger = LoggerFactory.getLogger(SQSCallback.class); private final AbstractAsyncProducer.CallbackCompleter cc; @@ -63,23 +69,23 @@ public SQSCallback(AbstractAsyncProducer.CallbackCompleter cc, Position position } @Override - public void onError(Exception t) { - logger.error(t.getClass().getSimpleName() + " @ " + position + " -- "); - 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(SendMessageResponse response, Throwable t) { + if (t != null) { + logger.error(t.getClass().getSimpleName() + " @ " + position + " -- "); + 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(SendMessageRequest request, SendMessageResult result) { if (logger.isDebugEnabled()) { logger.debug("-> Message id:{}, sequence number:{} {} {}", - result.getMessageId(), result.getSequenceNumber(), json, position); + response.messageId(), response.sequenceNumber(), json, position); } cc.markCompleted(); } diff --git a/src/test/java/com/zendesk/maxwell/producer/KinesisCallbackTest.java b/src/test/java/com/zendesk/maxwell/producer/KinesisCallbackTest.java index 555837dbf..1470ed0aa 100644 --- a/src/test/java/com/zendesk/maxwell/producer/KinesisCallbackTest.java +++ b/src/test/java/com/zendesk/maxwell/producer/KinesisCallbackTest.java @@ -2,7 +2,7 @@ import com.codahale.metrics.Counter; import com.codahale.metrics.Meter; -import com.amazonaws.services.kinesis.producer.IrrecoverableError; +import software.amazon.kinesis.producer.IrrecoverableError; import com.zendesk.maxwell.MaxwellConfig; import com.zendesk.maxwell.MaxwellContext; import com.zendesk.maxwell.replication.BinlogPosition; diff --git a/src/test/java/com/zendesk/maxwell/producer/MaxwellSNSProducerTest.java b/src/test/java/com/zendesk/maxwell/producer/MaxwellSNSProducerTest.java index 033c85e11..9ff1b84ed 100644 --- a/src/test/java/com/zendesk/maxwell/producer/MaxwellSNSProducerTest.java +++ b/src/test/java/com/zendesk/maxwell/producer/MaxwellSNSProducerTest.java @@ -1,12 +1,5 @@ package com.zendesk.maxwell.producer; -import com.amazonaws.services.sns.AmazonSNS; -import com.amazonaws.services.sns.AmazonSNSAsync; -import com.amazonaws.services.sns.AmazonSNSAsyncClient; -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.MaxwellConfig; import com.zendesk.maxwell.MaxwellContext; import com.zendesk.maxwell.monitoring.NoOpMetrics; @@ -22,21 +15,29 @@ import org.mockito.Captor; import org.mockito.Mockito; import org.mockito.MockitoAnnotations; +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.util.ArrayList; import java.util.Map; -import java.util.concurrent.Future; +import java.util.concurrent.CompletableFuture; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.when; -import static org.mockito.Mockito.*; + public class MaxwellSNSProducerTest { private static final long TIMESTAMP_MILLISECONDS = 1496712943447L; private static final String TOPIC = "topic"; private static final String FIFO_TOPIC = "topic.fifo"; private static final Position POSITION = new Position(new BinlogPosition(1L, "binlog-0001"), 0L); - private static final Future mockedFuture = Mockito.mock(Future.class); + private static final CompletableFuture mockedFuture = + CompletableFuture.completedFuture(PublishResponse.builder().messageId("mid").build()); + @Rule public final EnvironmentVariables environmentVariables = new EnvironmentVariables(); @@ -55,7 +56,7 @@ public void beforeEach() { @Test public void publishesRecord() throws Exception { - AmazonSNSAsyncClient client = Mockito.mock(AmazonSNSAsyncClient.class); + SnsAsyncClient client = Mockito.mock(SnsAsyncClient.class); MaxwellContext context = mock(MaxwellContext.class); when(context.getConfig()).thenReturn(new MaxwellConfig()); when(context.getMetrics()).thenReturn(new NoOpMetrics()); @@ -63,16 +64,16 @@ public void publishesRecord() throws Exception { producer.setClient(client); String payload = rowMap.toJSON(); AbstractAsyncProducer.CallbackCompleter cc = mock(AbstractAsyncProducer.CallbackCompleter.class); - when(client.publishAsync(any())).thenReturn(mockedFuture); + when(client.publish(any(PublishRequest.class))).thenReturn(mockedFuture); producer.sendAsync(rowMap, cc); - Mockito.verify(client, times(1)).publishAsync(arguments.capture(), any()); - Assert.assertEquals(arguments.getValue().getTopicArn(), TOPIC); - Assert.assertEquals(arguments.getValue().getMessage(), payload); + Mockito.verify(client, times(1)).publish(arguments.capture()); + Assert.assertEquals(arguments.getValue().topicArn(), TOPIC); + Assert.assertEquals(arguments.getValue().message(), payload); } @Test public void setsMessageAttributes() throws Exception { - AmazonSNSAsyncClient client = Mockito.mock(AmazonSNSAsyncClient.class); + SnsAsyncClient client = Mockito.mock(SnsAsyncClient.class); MaxwellContext context = mock(MaxwellContext.class); MaxwellConfig config = new MaxwellConfig(); config.snsAttrs = "database,table"; @@ -81,29 +82,28 @@ public void setsMessageAttributes() throws Exception { MaxwellSNSProducer producer = new MaxwellSNSProducer(context, TOPIC, "", ""); producer.setClient(client); AbstractAsyncProducer.CallbackCompleter cc = mock(AbstractAsyncProducer.CallbackCompleter.class); - when(client.publishAsync(any())).thenReturn(mockedFuture); + when(client.publish(any(PublishRequest.class))).thenReturn(mockedFuture); producer.sendAsync(rowMap, cc); - Mockito.verify(client, times(1)).publishAsync(arguments.capture(), any()); - Map attributes = arguments.getValue().getMessageAttributes(); + Mockito.verify(client, times(1)).publish(arguments.capture()); + Map attributes = arguments.getValue().messageAttributes(); Assert.assertNotNull(attributes.getOrDefault("table", null)); - Assert.assertEquals("MyTable", attributes.get("table").getStringValue()); + Assert.assertEquals("MyTable", attributes.get("table").stringValue()); Assert.assertNotNull(attributes.getOrDefault("database", null)); - Assert.assertEquals("MyDatabase", attributes.get("database").getStringValue()); + Assert.assertEquals("MyDatabase", attributes.get("database").stringValue()); } @Test public void ensureMessageGroupIdOnFifo() throws Exception { - AmazonSNSAsyncClient client = Mockito.mock(AmazonSNSAsyncClient.class); + SnsAsyncClient client = Mockito.mock(SnsAsyncClient.class); MaxwellContext context = mock(MaxwellContext.class); when(context.getConfig()).thenReturn(new MaxwellConfig()); when(context.getMetrics()).thenReturn(new NoOpMetrics()); MaxwellSNSProducer producer = new MaxwellSNSProducer(context, FIFO_TOPIC, "", ""); producer.setClient(client); AbstractAsyncProducer.CallbackCompleter cc = mock(AbstractAsyncProducer.CallbackCompleter.class); - when(client.publishAsync(any())).thenReturn(mockedFuture); + when(client.publish(any(PublishRequest.class))).thenReturn(mockedFuture); producer.sendAsync(rowMap, cc); - Mockito.verify(client, times(1)).publishAsync(arguments.capture(), any()); - Map attributes = arguments.getValue().getMessageAttributes(); - Assert.assertEquals("MyDatabase", arguments.getValue().getMessageGroupId()); + Mockito.verify(client, times(1)).publish(arguments.capture()); + Assert.assertEquals("MyDatabase", arguments.getValue().messageGroupId()); } }