fixed

sunxiaojian · Jun 26, 2024 · 8b54215 · 8b54215
1 parent 54b4672
commit 8b54215
Show file tree

Hide file tree

Showing 4 changed files with 7 additions and 11 deletions.
diff --git a/...ain/java/org/apache/paimon/flink/action/cdc/kafka/KafkaKeyValueDeserializationSchema.java b/...ain/java/org/apache/paimon/flink/action/cdc/kafka/KafkaKeyValueDeserializationSchema.java
@@ -61,10 +61,8 @@ public CdcSourceRecord deserialize(ConsumerRecord<byte[], byte[]> record) throws
                     record.key() != null
                             ? objectMapper.readValue(record.key(), JsonNode.class)
                             : null;
-            CdcSourceRecord sourceRecord =
-                    new CdcSourceRecord(
-                            key, objectMapper.readValue(record.value(), JsonNode.class));
-            return sourceRecord;
+            JsonNode value = objectMapper.readValue(record.value(), JsonNode.class);
+            return new CdcSourceRecord(key, value);
         } catch (Exception e) {
             LOG.error("Invalid Json:\n{}", new String(record.value()));
             throw e;

diff --git a/...n/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarKeyValueDeserializationSchema.java b/...n/java/org/apache/paimon/flink/action/cdc/pulsar/PulsarKeyValueDeserializationSchema.java
@@ -61,10 +61,8 @@ public CdcSourceRecord deserialize(Message<byte[]> message) throws IOException {
                     message.getKey() != null
                             ? objectMapper.readValue(message.getKey(), JsonNode.class)
                             : null;
-            CdcSourceRecord record =
-                    new CdcSourceRecord(
-                            key, objectMapper.readValue(message.getValue(), JsonNode.class));
-            return record;
+            JsonNode value = objectMapper.readValue(message.getValue(), JsonNode.class);
+            return new CdcSourceRecord(key, value);
         } catch (Exception e) {
             LOG.error("Invalid Json:\n{} \n{}", message.getKey(), new String(message.getValue()));
             throw e;

diff --git a/...ink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionITCaseBase.java b/...ink-cdc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaActionITCaseBase.java
@@ -351,7 +351,7 @@ private void send(String topic, String record, boolean wait) {
         }
     }
 
-    void writeRecordsToKafka(String topic, boolean wait, String resourceDirFormat)
+    void writeRecordsToKafkaWithKey(String topic, boolean wait, String resourceDirFormat)
             throws Exception {
         URL url =
                 KafkaCanalSyncTableActionITCase.class

diff --git a/...dc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableActionITCase.java b/...dc/src/test/java/org/apache/paimon/flink/action/cdc/kafka/KafkaSyncTableActionITCase.java
@@ -572,7 +572,7 @@ public void testSchemaIncludeRecord(String format) throws Exception {
     public void testRecordWithPrimaryKeys(String format) throws Exception {
         String topic = "no_schema_include_with_primary_keys";
         createTestTopic(topic, 1, 1);
-        writeRecordsToKafka(
+        writeRecordsToKafkaWithKey(
                 topic, false, "kafka/debezium/table/schema/primarykeys/debezium-data-1.txt");
         Map<String, String> kafkaConfig = getBasicKafkaConfig();
         kafkaConfig.put(VALUE_FORMAT.key(), format + "-json");
@@ -603,7 +603,7 @@ public void testRecordWithPrimaryKeys(String format) throws Exception {
     public void testSchemaIncludeRecordWithPrimaryKeys(String format) throws Exception {
         String topic = "schema_include_with_primary_keys";
         createTestTopic(topic, 1, 1);
-        writeRecordsToKafka(
+        writeRecordsToKafkaWithKey(
                 topic,
                 false,
                 "kafka/debezium/table/schema/primarykeys/debezium-data-with-schema-1.txt");