[FLINK-39696][runtime] Use source partition for distributed flush events (#4400)

Author: taoran92

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaOperator.java

@@ -195,11 +195,16 @@ public void processElement(StreamRecord<PartitioningEvent> streamRecord) throws
 
     private void requestSchemaChange(
             TableId sourceTableId, SchemaChangeRequest schemaChangeRequest) {
-        LOG.info("{}> Sent FlushEvent to downstream...", subTaskId);
+        final int sourcePartition = schemaChangeRequest.getSourceSubTaskId();
+
+        LOG.info(
+                "{}> Sent FlushEvent to downstream for source partition {}...",
+                subTaskId,
+                sourcePartition);
         output.collect(
                 new StreamRecord<>(
                         new FlushEvent(
-                                subTaskId,
+                                sourcePartition,
                                 tableIdRouter.route(sourceTableId),
                                 schemaChangeRequest.getSchemaChangeEvent().getType())));
 

flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/schema/distributed/SchemaEvolveTest.java

@@ -382,6 +382,37 @@ void testExceptionSchemaEvolution() throws Exception {
                         "Unexpected schema change events occurred in EXCEPTION mode. Job will fail now.");
     }
 
+    @Test
+    void testFlushEventUsesSourcePartitionInsteadOfSchemaOperatorSubtask() throws Exception {
+        CreateTableEvent createTableEvent = new CreateTableEvent(TABLE_ID, INITIAL_SCHEMA);
+
+        Assertions.assertThat(
+                        runInHarness(
+                                () ->
+                                        new SchemaOperator(
+                                                ROUTING_RULES,
+                                                RouteMode.ALL_MATCH,
+                                                Duration.ofMinutes(3),
+                                                SchemaChangeBehavior.LENIENT,
+                                                "UTC"),
+                                (op) ->
+                                        new DistributedEventOperatorTestHarness<>(
+                                                op,
+                                                20,
+                                                1,
+                                                Duration.ofSeconds(3),
+                                                Duration.ofMinutes(3)),
+                                (operator, harness) ->
+                                        operator.processElement(wrap(createTableEvent, 0, 1))))
+                .map(StreamRecord::getValue)
+                .first()
+                .isEqualTo(
+                        new FlushEvent(
+                                0,
+                                Collections.singletonList(TABLE_ID),
+                                createTableEvent.getType()));
+    }
+
     protected static <
                     OP extends AbstractStreamOperatorAdapter<E>,
                     E extends Event,

flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/testutils/operators/DistributedEventOperatorTestHarness.java

@@ -81,15 +81,26 @@ public class DistributedEventOperatorTestHarness<
     private final TestingSchemaRegistryGateway schemaRegistryGateway;
     private final LinkedList<StreamRecord<E>> outputRecords = new LinkedList<>();
     private final MockedOperatorCoordinatorContext mockedContext;
+    private final int subtaskIndex;
 
     public DistributedEventOperatorTestHarness(OP operator, int numOutputs) {
-        this(operator, numOutputs, Duration.ofSeconds(3), Duration.ofMinutes(3));
+        this(operator, numOutputs, 0, Duration.ofSeconds(3), Duration.ofMinutes(3));
     }
 
     public DistributedEventOperatorTestHarness(
             OP operator, int numOutputs, Duration applyDuration, Duration rpcTimeout) {
+        this(operator, numOutputs, 0, applyDuration, rpcTimeout);
+    }
+
+    public DistributedEventOperatorTestHarness(
+            OP operator,
+            int numOutputs,
+            int subtaskIndex,
+            Duration applyDuration,
+            Duration rpcTimeout) {
         this.operator = operator;
         this.numOutputs = numOutputs;
+        this.subtaskIndex = subtaskIndex;
         this.mockedContext =
                 new MockedOperatorCoordinatorContext(
                         SCHEMA_OPERATOR_ID, Thread.currentThread().getContextClassLoader());
@@ -160,7 +171,7 @@ public void close() throws Exception {
 
     private void initializeOperator() throws Exception {
         operator.setup(
-                new MockStreamTask(schemaRegistryGateway),
+                new MockStreamTask(schemaRegistryGateway, subtaskIndex),
                 new MockStreamConfig(new Configuration(), numOutputs),
                 new EventCollectingOutput<>(outputRecords, schemaRegistryGateway));
         schemaRegistryGateway.sendOperatorEventToCoordinator(
@@ -227,9 +238,10 @@ public void close() {}
     }
 
     private static class MockStreamTask extends StreamTask<Event, AbstractStreamOperator<Event>> {
-        protected MockStreamTask(TestingSchemaRegistryGateway schemaRegistryGateway)
+        protected MockStreamTask(
+                TestingSchemaRegistryGateway schemaRegistryGateway, int subtaskIndex)
                 throws Exception {
-            super(new SchemaRegistryCoordinatingEnvironment(schemaRegistryGateway));
+            super(new SchemaRegistryCoordinatingEnvironment(schemaRegistryGateway, subtaskIndex));
         }
 
         @Override
@@ -240,7 +252,8 @@ private static class SchemaRegistryCoordinatingEnvironment extends DummyEnvironm
         private final TestingSchemaRegistryGateway schemaRegistryGateway;
 
         public SchemaRegistryCoordinatingEnvironment(
-                TestingSchemaRegistryGateway schemaRegistryGateway) {
+                TestingSchemaRegistryGateway schemaRegistryGateway, int subtaskIndex) {
+            super("test-task", 2, subtaskIndex, 2);
             this.schemaRegistryGateway = schemaRegistryGateway;
         }