Make SNS/SQS procuders to respect outputConfig

DataEngDev · Apr 4, 2022 · f977051 · f977051
1 parent c584bb2
commit f977051
Show file tree

Hide file tree

Showing 2 changed files with 2 additions and 2 deletions.
diff --git a/src/main/java/com/zendesk/maxwell/producer/MaxwellSNSProducer.java b/src/main/java/com/zendesk/maxwell/producer/MaxwellSNSProducer.java
@@ -35,7 +35,7 @@ public void setClient(AmazonSNSAsync client) {
 
 	@Override
 	public void sendAsync(RowMap r, CallbackCompleter cc) throws Exception {
-		String value = r.toJSON();
+		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<>();

diff --git a/src/main/java/com/zendesk/maxwell/producer/MaxwellSQSProducer.java b/src/main/java/com/zendesk/maxwell/producer/MaxwellSQSProducer.java
@@ -26,7 +26,7 @@ public MaxwellSQSProducer(MaxwellContext context, String queueUri) {
 
 	@Override
 	public void sendAsync(RowMap r, CallbackCompleter cc) throws Exception {
-		String value = r.toJSON();
+		String value = r.toJSON(outputConfig);
 		SendMessageRequest messageRequest = new SendMessageRequest(queueUri, value);
 		if ( queueUri.endsWith(".fifo")) {
 			messageRequest.setMessageGroupId(r.getDatabase());