This is an automated email from the ASF dual-hosted git repository.
lvyanquan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new f830a6877 [FLINK-41317][pipeline-connector/kafka] Fix NPE when
serializing REPLACE events without before image (#4610)
f830a6877 is described below
commit f830a687715fff53a6e3501c23904bdc227bdd32
Author: κΉλν <[email protected]>
AuthorDate: Wed Nov 4 10:34:55 2026 +0900
[FLINK-41317][pipeline-connector/kafka] Fix NPE when serializing REPLACE
events without before image (#4610)
---
.../json/canal/CanalJsonSerializationSchema.java | 18 +++++++++++++++++-
.../debezium/DebeziumJsonSerializationSchema.java | 21 ++++++++++++++++++++-
.../canal/CanalJsonSerializationSchemaTest.java | 14 ++++++++++++++
.../DebeziumJsonSerializationSchemaTest.java | 14 ++++++++++++++
4 files changed, 65 insertions(+), 2 deletions(-)
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonSerializationSchema.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonSerializationSchema.java
index a9a0209a6..d39206402 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonSerializationSchema.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonSerializationSchema.java
@@ -187,7 +187,6 @@ public class CanalJsonSerializationSchema implements
SerializationSchema<Event>
.getSerializationSchema()
.serialize(reuseGenericRowData);
case UPDATE:
- case REPLACE:
reuseGenericRowData.setField(
0,
new GenericArrayData(
@@ -211,6 +210,23 @@ public class CanalJsonSerializationSchema implements
SerializationSchema<Event>
.get(dataChangeEvent.tableId())
.getSerializationSchema()
.serialize(reuseGenericRowData);
+ case REPLACE:
+ // REPLACE events carry only the after image, so "old" is
written as null.
+ reuseGenericRowData.setField(0, null);
+ reuseGenericRowData.setField(
+ 1,
+ new GenericArrayData(
+ new RowData[] {
+ jsonSerializers
+ .get(dataChangeEvent.tableId())
+ .getRowDataFromRecordData(
+
dataChangeEvent.after(), false)
+ }));
+ reuseGenericRowData.setField(2, OP_UPDATE);
+ return jsonSerializers
+ .get(dataChangeEvent.tableId())
+ .getSerializationSchema()
+ .serialize(reuseGenericRowData);
default:
throw new UnsupportedOperationException(
format(
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/debezium/DebeziumJsonSerializationSchema.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/debezium/DebeziumJsonSerializationSchema.java
index cbda93dfd..10275333b 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/debezium/DebeziumJsonSerializationSchema.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/main/java/org/apache/flink/cdc/connectors/kafka/json/debezium/DebeziumJsonSerializationSchema.java
@@ -208,9 +208,11 @@ public class DebeziumJsonSerializationSchema implements
SerializationSchema<Even
converter = this::convertDeleteEventToRowData;
break;
case UPDATE:
- case REPLACE:
converter = this::convertUpdateEventToRowData;
break;
+ case REPLACE:
+ converter = this::convertReplaceEventToRowData;
+ break;
default:
throw new UnsupportedOperationException(
format(
@@ -476,6 +478,23 @@ public class DebeziumJsonSerializationSchema implements
SerializationSchema<Even
StringData.fromString(dataChangeEvent.tableId().getTableName())));
}
+ private void convertReplaceEventToRowData(
+ DataChangeEvent dataChangeEvent, GenericRowData genericRowData) {
+ // REPLACE events carry only the after image, so "before" is written
as null.
+ genericRowData.setField(BEFORE.getPosition(), null);
+ genericRowData.setField(
+ AFTER.getPosition(),
+ jsonSerializers
+ .get(dataChangeEvent.tableId())
+ .getRowDataFromRecordData(dataChangeEvent.after(),
false));
+ genericRowData.setField(OPERATION.getPosition(), OP_UPDATE);
+ genericRowData.setField(
+ SOURCE.getPosition(),
+ GenericRowData.of(
+
StringData.fromString(dataChangeEvent.tableId().getSchemaName()),
+
StringData.fromString(dataChangeEvent.tableId().getTableName())));
+ }
+
/**
* Refer to <a
*
href="https://debezium.io/documentation/reference/1.9/connectors/mysql.html";>Debezium
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonSerializationSchemaTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonSerializationSchemaTest.java
index dc2b71d43..cb78f8fbd 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonSerializationSchemaTest.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/canal/CanalJsonSerializationSchemaTest.java
@@ -133,6 +133,20 @@ class CanalJsonSerializationSchemaTest {
"{\"old\":[{\"col1\":\"1\",\"col2\":\"1\"}],\"data\":[{\"col1\":\"1\",\"col2\":\"x\"}],\"type\":\"UPDATE\",\"database\":\"default_schema\",\"table\":\"table1\",\"pkNames\":[\"col1\"]}");
actual = mapper.readTree(serializationSchema.serialize(updateEvent));
Assertions.assertThat(actual).isEqualTo(expected);
+
+ DataChangeEvent replaceEvent =
+ DataChangeEvent.replaceEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("y")
+ }));
+ expected =
+ mapper.readTree(
+
"{\"old\":null,\"data\":[{\"col1\":\"1\",\"col2\":\"y\"}],\"type\":\"UPDATE\",\"database\":\"default_schema\",\"table\":\"table1\",\"pkNames\":[\"col1\"]}");
+ actual = mapper.readTree(serializationSchema.serialize(replaceEvent));
+ Assertions.assertThat(actual).isEqualTo(expected);
}
@Test
diff --git
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/debezium/DebeziumJsonSerializationSchemaTest.java
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/debezium/DebeziumJsonSerializationSchemaTest.java
index 8ad302560..0be588305 100644
---
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/debezium/DebeziumJsonSerializationSchemaTest.java
+++
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-kafka/src/test/java/org/apache/flink/cdc/connectors/kafka/json/debezium/DebeziumJsonSerializationSchemaTest.java
@@ -138,6 +138,20 @@ class DebeziumJsonSerializationSchemaTest {
"{\"before\":{\"col1\":\"1\",\"col2\":\"1\"},\"after\":{\"col1\":\"1\",\"col2\":\"x\"},\"op\":\"u\",\"source\":{\"db\":\"default_schema\",\"table\":\"table1\"}}");
actual = mapper.readTree(serializationSchema.serialize(updateEvent));
assertThat(actual).isEqualTo(expected);
+
+ DataChangeEvent replaceEvent =
+ DataChangeEvent.replaceEvent(
+ TABLE_1,
+ generator.generate(
+ new Object[] {
+ BinaryStringData.fromString("1"),
+ BinaryStringData.fromString("y")
+ }));
+ expected =
+ mapper.readTree(
+
"{\"before\":null,\"after\":{\"col1\":\"1\",\"col2\":\"y\"},\"op\":\"u\",\"source\":{\"db\":\"default_schema\",\"table\":\"table1\"}}");
+ actual = mapper.readTree(serializationSchema.serialize(replaceEvent));
+ assertThat(actual).isEqualTo(expected);
}
@Test
