[cdc] Fix duplicate schema IDs when multiple parallel updates (apache…

…#2797)
lmagic233 · Feb 2, 2024 · 0f117d4 · 0f117d4
1 parent d099a79
commit 0f117d4
Showing 1 changed file with 1 addition and 0 deletions.
diff --git a/...k-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java b/...k-cdc/src/main/java/org/apache/paimon/flink/sink/cdc/FlinkCdcSyncDatabaseSinkBuilder.java
@@ -149,6 +149,7 @@ private void buildCombinedCdcSink() {
         SingleOutputStreamOperatorUtils.getSideOutput(
                         parsed,
                         CdcDynamicTableParsingProcessFunction.DYNAMIC_SCHEMA_CHANGE_OUTPUT_TAG)
+                .keyBy(t -> t.f0)
                 .process(new MultiTableUpdatedDataFieldsProcessFunction(catalogLoader))
                 .name("Schema Evolution");