0005948: KafkaDataWriter by ROW method has a trailing comma

JumpMind · Aug 8, 2023 · 14ae876 · 14ae876
1 parent 56312b4
commit 14ae876
Showing 1 changed file with 4 additions and 1 deletion.
diff --git a/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/writer/KafkaWriter.java b/symmetric-io/src/main/java/org/jumpmind/symmetric/io/data/writer/KafkaWriter.java
@@ -221,6 +221,9 @@ protected int execute(CsvData data, String[] values) {
         }
         List<ProducerRecord<String, Object>> kafkaDataList = kafkaDataMap.get(kafkaDataKey);
         if (outputFormat.equals(KAFKA_FORMAT_JSON)) {
+            if(!kafkaDataList.isEmpty() && !messageBy.equals(KAFKA_MESSAGE_BY_ROW)) {
+               kafkaText.append(", ");
+            }
             kafkaText.append("{\"").append(table.getName()).append("\": {").append("\"eventType\": \"" + data.getDataEventType() + "\",")
                     .append("\"data\": { ");
             // Let Gson escape the json values
@@ -232,7 +235,7 @@ protected int execute(CsvData data, String[] values) {
                     kafkaText.append(",");
                 }
             }
-            kafkaText.append(" } } }, ");
+            kafkaText.append(" } } }");
         } else if (outputFormat.equals(KAFKA_FORMAT_CSV)) {
             // Quote every non-null field, escape quote character by
             // doubling the quote character