[hotfix] Fix pre-partition broadcasting failure due to shallow-copying

apache · Aug 25, 2024 · bdf051b · bdf051b
1 parent 69ca252
commit bdf051b
Showing 1 changed file with 5 additions and 1 deletion.
diff --git a/...runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/PrePartitionOperator.java b/...runtime/src/main/java/org/apache/flink/cdc/runtime/partitioning/PrePartitionOperator.java
@@ -28,6 +28,7 @@
 import org.apache.flink.cdc.common.schema.Schema;
 import org.apache.flink.cdc.runtime.operators.schema.SchemaOperator;
 import org.apache.flink.cdc.runtime.operators.sink.SchemaEvolutionClient;
+import org.apache.flink.cdc.runtime.serializer.event.EventSerializer;
 import org.apache.flink.runtime.jobgraph.OperatorID;
 import org.apache.flink.runtime.jobgraph.tasks.TaskOperatorEventGateway;
 import org.apache.flink.streaming.api.operators.AbstractStreamOperator;
@@ -108,7 +109,10 @@ private void partitionBy(DataChangeEvent dataChangeEvent) throws Exception {
 
     private void broadcastEvent(Event toBroadcast) {
         for (int i = 0; i < downstreamParallelism; i++) {
-            output.collect(new StreamRecord<>(new PartitioningEvent(toBroadcast, i)));
+            // Deep-copying each event is required since downstream subTasks might run in the same
+            // JVM
+            Event copiedEvent = EventSerializer.INSTANCE.copy(toBroadcast);
+            output.collect(new StreamRecord<>(new PartitioningEvent(copiedEvent, i)));
         }
     }