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 43f36c9eb1 [hotfix][runtime] Fix unable to take __data_event_type__
from REPLACE events (#4531)
43f36c9eb1 is described below
commit 43f36c9eb140f61e8a9bf6183ae9a544ffd69d52
Author: yuxiqian <[email protected]>
AuthorDate: Thu Sep 17 11:41:02 2026 +0800
[hotfix][runtime] Fix unable to take __data_event_type__ from REPLACE
events (#4531)
---
.../flink/cdc/common/event/DataChangeEvent.java | 2 +
.../cdc/common/event/DataChangeEventTest.java | 47 ++++++++++++++++++++
.../transform/PostTransformOperatorTest.java | 50 ++++++++++++++++++++++
3 files changed, 99 insertions(+)
diff --git
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/DataChangeEvent.java
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/DataChangeEvent.java
index 92b2504aa6..9fd036f193 100644
---
a/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/DataChangeEvent.java
+++
b/flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/DataChangeEvent.java
@@ -87,6 +87,8 @@ public class DataChangeEvent implements ChangeEvent,
Serializable {
return isAfter ? "+U" : "-U";
case DELETE:
return "-D";
+ case REPLACE:
+ return "+U";
default:
throw new UnsupportedOperationException("Unknown operation
type: " + op);
}
diff --git
a/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/event/DataChangeEventTest.java
b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/event/DataChangeEventTest.java
new file mode 100644
index 0000000000..a3d6bbfb1a
--- /dev/null
+++
b/flink-cdc-common/src/test/java/org/apache/flink/cdc/common/event/DataChangeEventTest.java
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.common.event;
+
+import org.apache.flink.cdc.common.data.GenericRecordData;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Unit tests for {@link DataChangeEvent}. */
+class DataChangeEventTest {
+
+ private static final TableId TABLE_ID = TableId.tableId("schema", "table");
+ private static final GenericRecordData RECORD =
GenericRecordData.of("record");
+
+ @Test
+ void testOpTypeString() {
+ assertThat(DataChangeEvent.insertEvent(TABLE_ID,
RECORD).opTypeString(true))
+ .isEqualTo("+I");
+ assertThat(DataChangeEvent.updateEvent(TABLE_ID, RECORD,
RECORD).opTypeString(false))
+ .isEqualTo("-U");
+ assertThat(DataChangeEvent.updateEvent(TABLE_ID, RECORD,
RECORD).opTypeString(true))
+ .isEqualTo("+U");
+ assertThat(DataChangeEvent.deleteEvent(TABLE_ID,
RECORD).opTypeString(false))
+ .isEqualTo("-D");
+ assertThat(DataChangeEvent.replaceEvent(TABLE_ID,
RECORD).opTypeString(false))
+ .isEqualTo("+U");
+ assertThat(DataChangeEvent.replaceEvent(TABLE_ID,
RECORD).opTypeString(true))
+ .isEqualTo("+U");
+ }
+}
diff --git
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java
index 2e02b05a3b..d784c4a2bd 100644
---
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java
+++
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java
@@ -705,6 +705,56 @@ class PostTransformOperatorTest {
transformFunctionEventEventOperatorTestHarness.close();
}
+ @Test
+ void testReplaceEventMetadataTransform() throws Exception {
+ PostTransformOperator transform =
+ PostTransformOperator.newBuilder()
+ .addTransform(
+ METADATA_TABLEID.identifier(),
+ "*, __data_event_type__",
+ " __table_name__ = 'metadata_table' ")
+ .build();
+ RegularEventOperatorTestHarness<PostTransformOperator, Event>
+ transformFunctionEventEventOperatorTestHarness =
+ RegularEventOperatorTestHarness.with(transform, 1);
+ Schema expectedSchema =
+ Schema.newBuilder()
+ .physicalColumn("col1", DataTypes.STRING().notNull())
+ .physicalColumn("__data_event_type__",
DataTypes.STRING().notNull())
+ .primaryKey("col1")
+ .build();
+
+ transformFunctionEventEventOperatorTestHarness.open();
+
+ CreateTableEvent createTableEvent = new
CreateTableEvent(METADATA_TABLEID, METADATA_SCHEMA);
+ BinaryRecordDataGenerator recordDataGenerator =
+ new BinaryRecordDataGenerator(((RowType)
METADATA_SCHEMA.toRowDataType()));
+ BinaryRecordDataGenerator expectedRecordDataGenerator =
+ new BinaryRecordDataGenerator(((RowType)
expectedSchema.toRowDataType()));
+ DataChangeEvent replaceEvent =
+ DataChangeEvent.replaceEvent(
+ METADATA_TABLEID,
+ recordDataGenerator.generate(new Object[] {new
BinaryStringData("1")}));
+ DataChangeEvent replaceEventExpect =
+ DataChangeEvent.replaceEvent(
+ METADATA_TABLEID,
+ expectedRecordDataGenerator.generate(
+ new Object[] {
+ new BinaryStringData("1"), new
BinaryStringData("+U")
+ }));
+
+ transform.processElement(new StreamRecord<>(createTableEvent));
+ Assertions.assertThat(
+
transformFunctionEventEventOperatorTestHarness.getOutputRecords().poll())
+ .isEqualTo(
+ new StreamRecord<>(new
CreateTableEvent(METADATA_TABLEID, expectedSchema)));
+ transform.processElement(new StreamRecord<>(replaceEvent));
+ Assertions.assertThat(
+
transformFunctionEventEventOperatorTestHarness.getOutputRecords().poll())
+ .isEqualTo(new StreamRecord<>(replaceEventExpect));
+ transformFunctionEventEventOperatorTestHarness.close();
+ }
+
@Test
void testMetadataASTransform() throws Exception {
PostTransformOperator transform =
