[FLINK-36701][cdc-runtime] Add steps to get and emit schemaManager's latest evolvedSchema when SinkDataWriterOperator handles FlushEvent.

Author: Jzjsnow

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/event/FlushEvent.java

@@ -17,25 +17,46 @@
 
 package org.apache.flink.cdc.common.event;
 
+import java.util.Collections;
+import java.util.List;
 import java.util.Objects;
 
 /**
  * An {@link Event} from {@code SchemaOperator} to notify {@code DataSinkWriterOperator} that it
  * start flushing.
  */
 public class FlushEvent implements Event {
+    /** The schema changes from which table. */
+    private final List<TableId> tableIds;
 
     /** Which subTask ID this FlushEvent was initiated from. */
     private final int sourceSubTaskId;
 
+    /** Flag indicating whether the FlushEvent is sent before a create table event. */
+    private final Boolean isForCreateTableEvent;
+
     public FlushEvent(int sourceSubTaskId) {
+        this(sourceSubTaskId, Collections.emptyList(), false);
+    }
+
+    public FlushEvent(int sourceSubTaskId, List<TableId> tableIds, boolean isForCreateTableEvent) {
+        this.tableIds = tableIds;
         this.sourceSubTaskId = sourceSubTaskId;
+        this.isForCreateTableEvent = isForCreateTableEvent;
+    }
+
+    public List<TableId> getTableIds() {
+        return tableIds;
     }
 
     public int getSourceSubTaskId() {
         return sourceSubTaskId;
     }
 
+    public Boolean getIsForCreateTableEvent() {
+        return isForCreateTableEvent;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -45,7 +66,9 @@ public boolean equals(Object o) {
             return false;
         }
         FlushEvent that = (FlushEvent) o;
-        return sourceSubTaskId == that.sourceSubTaskId;
+        return sourceSubTaskId == that.sourceSubTaskId
+                && Objects.equals(tableIds, that.tableIds)
+                && Objects.equals(isForCreateTableEvent, that.isForCreateTableEvent);
     }
 
     @Override

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

@@ -22,6 +22,7 @@
 import org.apache.flink.cdc.common.event.Event;
 import org.apache.flink.cdc.common.event.FlushEvent;
 import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.SchemaChangeEventType;
 import org.apache.flink.cdc.common.event.TableId;
 import org.apache.flink.cdc.common.exceptions.SchemaEvolveException;
 import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
@@ -150,6 +151,7 @@ public void processElement(StreamRecord<PartitioningEvent> streamRecord) throws
 
             // Then, notify this information to the coordinator
             requestSchemaChange(
+                    tableId,
                     new SchemaChangeRequest(sourcePartition, subTaskId, schemaChangeEvent));
             schemaOperatorMetrics.increaseFinishedSchemaChangeEvents(1);
         } else if (event instanceof DataChangeEvent) {
@@ -188,9 +190,16 @@ public void processElement(StreamRecord<PartitioningEvent> streamRecord) throws
         }
     }
 
-    private void requestSchemaChange(SchemaChangeRequest schemaChangeRequest) {
+    private void requestSchemaChange(
+            TableId sourceTableId, SchemaChangeRequest schemaChangeRequest) {
         LOG.info("{}> Sent FlushEvent to downstream...", subTaskId);
-        output.collect(new StreamRecord<>(new FlushEvent(subTaskId)));
+        output.collect(
+                new StreamRecord<>(
+                        new FlushEvent(
+                                subTaskId,
+                                tableIdRouter.route(sourceTableId),
+                                schemaChangeRequest.getSchemaChangeEvent().getType()
+                                        == SchemaChangeEventType.CREATE_TABLE)));
 
         LOG.info("{}> Sending evolve request...", subTaskId);
         SchemaChangeResponse response = sendRequestToCoordinator(schemaChangeRequest);

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

@@ -23,6 +23,7 @@
 import org.apache.flink.cdc.common.event.Event;
 import org.apache.flink.cdc.common.event.FlushEvent;
 import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.SchemaChangeEventType;
 import org.apache.flink.cdc.common.event.TableId;
 import org.apache.flink.cdc.common.pipeline.SchemaChangeBehavior;
 import org.apache.flink.cdc.common.route.RouteRule;
@@ -164,8 +165,14 @@ private void handleSchemaChangeEvent(SchemaChangeEvent originalEvent) throws Exc
         schemaOperatorMetrics.increaseSchemaChangeEvents(1);
 
         // First, send FlushEvent or it might be blocked later
+        List<TableId> sinkTables = router.route(tableId);
         LOG.info("{}> Sending the FlushEvent.", subTaskId);
-        output.collect(new StreamRecord<>(new FlushEvent(subTaskId)));
+        output.collect(
+                new StreamRecord<>(
+                        new FlushEvent(
+                                subTaskId,
+                                sinkTables,
+                                originalEvent.getType() == SchemaChangeEventType.CREATE_TABLE)));
 
         // Then, queue to request schema change to SchemaCoordinator.
         SchemaChangeResponse response = requestSchemaChange(tableId, originalEvent);

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkFunctionOperator.java

@@ -112,6 +112,20 @@ public void processElement(StreamRecord<Event> element) throws Exception {
     // ----------------------------- Helper functions -------------------------------
     private void handleFlushEvent(FlushEvent event) throws Exception {
         userFunction.finish();
+        if (!event.getIsForCreateTableEvent()) {
+            event.getTableIds().stream()
+                    .filter(tableId -> !processedTableIds.contains(tableId))
+                    .forEach(
+                            tableId -> {
+                                LOG.info("Table {} has not been processed", tableId);
+                                try {
+                                    emitLatestSchema(tableId);
+                                } catch (Exception e) {
+                                    throw new RuntimeException(e);
+                                }
+                                processedTableIds.add(tableId);
+                            });
+        }
         schemaEvolutionClient.notifyFlushSuccess(
                 getRuntimeContext().getIndexOfThisSubtask(), event.getSourceSubTaskId());
     }

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/operators/sink/DataSinkWriterOperator.java

@@ -198,6 +198,20 @@ public void endInput() throws Exception {
 
     private void handleFlushEvent(FlushEvent event) throws Exception {
         copySinkWriter.flush(false);
+        if (!event.getIsForCreateTableEvent()) {
+            event.getTableIds().stream()
+                    .filter(tableId -> !processedTableIds.contains(tableId))
+                    .forEach(
+                            tableId -> {
+                                LOG.info("Table {} has not been processed", tableId);
+                                try {
+                                    emitLatestSchema(tableId);
+                                } catch (Exception e) {
+                                    throw new RuntimeException(e);
+                                }
+                                processedTableIds.add(tableId);
+                            });
+        }
         schemaEvolutionClient.notifyFlushSuccess(
                 getRuntimeContext().getIndexOfThisSubtask(), event.getSourceSubTaskId());
     }

flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/serializer/event/EventSerializer.java

@@ -24,7 +24,10 @@
 import org.apache.flink.cdc.common.event.Event;
 import org.apache.flink.cdc.common.event.FlushEvent;
 import org.apache.flink.cdc.common.event.SchemaChangeEvent;
+import org.apache.flink.cdc.common.event.TableId;
+import org.apache.flink.cdc.runtime.serializer.BooleanSerializer;
 import org.apache.flink.cdc.runtime.serializer.EnumSerializer;
+import org.apache.flink.cdc.runtime.serializer.ListSerializer;
 import org.apache.flink.cdc.runtime.serializer.TableIdSerializer;
 import org.apache.flink.cdc.runtime.serializer.TypeSerializerSingleton;
 import org.apache.flink.core.memory.DataInputView;
@@ -42,11 +45,13 @@ public final class EventSerializer extends TypeSerializerSingleton<Event> {
 
     private final SchemaChangeEventSerializer schemaChangeEventSerializer =
             SchemaChangeEventSerializer.INSTANCE;
-    private final TableIdSerializer tableIdSerializer = TableIdSerializer.INSTANCE;
+    private final ListSerializer<TableId> listSerializer =
+            new ListSerializer<>(TableIdSerializer.INSTANCE);
     private final EnumSerializer<EventClass> enumSerializer =
             new EnumSerializer<>(EventClass.class);
     private final TypeSerializer<DataChangeEvent> dataChangeEventSerializer =
             DataChangeEventSerializer.INSTANCE;
+    private final BooleanSerializer booleanSerializer = BooleanSerializer.INSTANCE;
 
     @Override
     public boolean isImmutableType() {
@@ -62,7 +67,10 @@ public Event createInstance() {
     public Event copy(Event from) {
         if (from instanceof FlushEvent) {
             FlushEvent flushEvent = (FlushEvent) from;
-            return new FlushEvent(((FlushEvent) from).getSourceSubTaskId());
+            return new FlushEvent(
+                    flushEvent.getSourceSubTaskId(),
+                    listSerializer.copy(((FlushEvent) from).getTableIds()),
+                    booleanSerializer.copy(flushEvent.getIsForCreateTableEvent()));
         } else if (from instanceof SchemaChangeEvent) {
             return schemaChangeEventSerializer.copy((SchemaChangeEvent) from);
         } else if (from instanceof DataChangeEvent) {
@@ -86,6 +94,8 @@ public void serialize(Event record, DataOutputView target) throws IOException {
         if (record instanceof FlushEvent) {
             enumSerializer.serialize(EventClass.FLUSH_EVENT, target);
             target.writeInt(((FlushEvent) record).getSourceSubTaskId());
+            listSerializer.serialize(((FlushEvent) record).getTableIds(), target);
+            booleanSerializer.serialize(((FlushEvent) record).getIsForCreateTableEvent(), target);
         } else if (record instanceof SchemaChangeEvent) {
             enumSerializer.serialize(EventClass.SCHEME_CHANGE_EVENT, target);
             schemaChangeEventSerializer.serialize((SchemaChangeEvent) record, target);
@@ -102,7 +112,10 @@ public Event deserialize(DataInputView source) throws IOException {
         EventClass eventClass = enumSerializer.deserialize(source);
         switch (eventClass) {
             case FLUSH_EVENT:
-                return new FlushEvent(source.readInt());
+                return new FlushEvent(
+                        source.readInt(),
+                        listSerializer.deserialize(source),
+                        booleanSerializer.deserialize(source));
             case DATA_CHANGE_EVENT:
                 return dataChangeEventSerializer.deserialize(source);
             case SCHEME_CHANGE_EVENT:

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

@@ -181,16 +181,16 @@ void testLenientSchemaEvolution() throws Exception {
                                 }))
                 .map(StreamRecord::getValue)
                 .containsExactly(
-                        new FlushEvent(0),
+                        new FlushEvent(0, Collections.singletonList(TABLE_ID), true),
                         createTableEvent,
                         genInsert(TABLE_ID, "ISFS", 1, "Alice", 17.1828f, "Hello"),
-                        new FlushEvent(0),
+                        new FlushEvent(0, Collections.singletonList(TABLE_ID), false),
                         addColumnEventAtLast,
                         genInsert(TABLE_ID, "ISFSB", 2, "Bob", 31.415926f, "Bye-bye", false),
-                        new FlushEvent(0),
+                        new FlushEvent(0, Collections.singletonList(TABLE_ID), false),
                         appendRenamedColumnAtLast,
                         genInsert(TABLE_ID, "ISFSBS", 3, "Cicada", 123.456f, null, true, "Ok"),
-                        new FlushEvent(0),
+                        new FlushEvent(0, Collections.singletonList(TABLE_ID), false),
                         alterColumnTypeEventWithBackfill,
                         genInsert(
                                 TABLE_ID,
@@ -201,11 +201,11 @@ void testLenientSchemaEvolution() throws Exception {
                                 null,
                                 false,
                                 "Nah"),
-                        new FlushEvent(0),
+                        new FlushEvent(0, Collections.singletonList(TABLE_ID), false),
                         genInsert(TABLE_ID, "ISDSBS", 5, "Eve", 1.414, null, true, null),
-                        new FlushEvent(0),
+                        new FlushEvent(0, Collections.singletonList(TABLE_ID), false),
                         genInsert(TABLE_ID, "ISDSBS", 6, "Ferris", 0.001, null, false, null),
-                        new FlushEvent(0));
+                        new FlushEvent(0, Collections.singletonList(TABLE_ID), false));
     }
 
     @Test
@@ -308,7 +308,7 @@ void testIgnoreSchemaEvolution() throws Exception {
                                 }))
                 .map(StreamRecord::getValue)
                 .containsExactly(
-                        new FlushEvent(0),
+                        new FlushEvent(0, Collections.singletonList(TABLE_ID), true),
                         createTableEvent,
                         genInsert(TABLE_ID, "ISFS", 1, "Alice", 17.1828f, "Hello"),
                         genInsert(TABLE_ID, "ISFS", 2, "Bob", 31.415926f, "Bye-bye"),