[Feature][Flink] Support CDC source schema evolution on Flink engine (#9867)

Author: CloverDew

seatunnel-api/src/main/java/org/apache/seatunnel/api/event/EventType.java

@@ -31,4 +31,5 @@ public enum EventType {
     LIFECYCLE_WRITER_CLOSE,
     READER_MESSAGE_DELAYED,
     JOB_STATUS,
+    SCHEMA_CHANGE_FLUSH,
 }

seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/SupportSchemaEvolutionSinkWriter.java

@@ -17,6 +17,8 @@
 
 package org.apache.seatunnel.api.sink;
 
+import org.apache.seatunnel.api.table.coordinator.SchemaCoordinator;
+import org.apache.seatunnel.api.table.schema.event.FlushEvent;
 import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
 
 import java.io.IOException;
@@ -30,4 +32,48 @@ public interface SupportSchemaEvolutionSinkWriter {
      * @throws IOException
      */
     void applySchemaChange(SchemaChangeEvent event) throws IOException;
+
+    /**
+     * handle FlushEvent propagated from upstream
+     *
+     * @param event
+     * @throws IOException
+     */
+    default void handleFlushEvent(FlushEvent event) throws IOException {
+        flushData();
+        sendFlushSuccessful(event);
+    }
+
+    /**
+     * send success event to coordinator upon successful flash
+     *
+     * @param event
+     * @throws IOException
+     */
+    default void sendFlushSuccessful(FlushEvent event) throws IOException {
+        SchemaCoordinator coordinator = getSchemaCoordinator();
+        if (coordinator == null && event != null && event.getJobId() != null) {
+            coordinator = SchemaCoordinator.getOrCreateInstance(event.getJobId());
+        }
+
+        if (coordinator != null) {
+            coordinator.notifyFlushSuccessful(event.getJobId(), event.tableIdentifier());
+        }
+    }
+
+    /**
+     * Get the schema coordinator instance for reporting flush completion
+     *
+     * @return the schema coordinator instance, or null if not available
+     */
+    default SchemaCoordinator getSchemaCoordinator() {
+        return null;
+    }
+
+    /**
+     * flush data to other system
+     *
+     * @throws IOException
+     */
+    default void flushData() throws IOException {}
 }

seatunnel-api/src/main/java/org/apache/seatunnel/api/sink/multitablesink/MultiTableSinkWriter.java

@@ -21,6 +21,7 @@
 import org.apache.seatunnel.api.sink.SinkWriter;
 import org.apache.seatunnel.api.sink.SupportMultiTableSinkWriter;
 import org.apache.seatunnel.api.sink.SupportSchemaEvolutionSinkWriter;
+import org.apache.seatunnel.api.table.schema.event.FlushEvent;
 import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
 import org.apache.seatunnel.api.tracing.MDCTracer;
@@ -175,8 +176,36 @@ public void applySchemaChange(SchemaChangeEvent event) throws IOException {
         }
     }
 
+    @Override
+    public void handleFlushEvent(FlushEvent event) throws IOException {
+        subSinkErrorCheck();
+        String targetTableId = event.tableIdentifier().toTablePath().getFullName();
+        for (int i = 0; i < sinkWritersWithIndex.size(); i++) {
+            for (Map.Entry<SinkIdentifier, SinkWriter<SeaTunnelRow, ?, ?>> sinkWriterEntry :
+                    sinkWritersWithIndex.get(i).entrySet()) {
+                if (sinkWriterEntry.getKey().getTableIdentifier().equals(targetTableId)) {
+                    synchronized (runnable.get(i)) {
+                        SinkWriter<SeaTunnelRow, ?, ?> sink = sinkWriterEntry.getValue();
+                        if (sink instanceof SupportSchemaEvolutionSinkWriter) {
+                            ((SupportSchemaEvolutionSinkWriter) sink).handleFlushEvent(event);
+                        }
+                    }
+                    return;
+                }
+            }
+        }
+    }
+
     @Override
     public void write(SeaTunnelRow element) throws IOException {
+        if (element != null && element.getOptions() != null) {
+            if (element.getOptions().containsKey("flush_event")
+                    || element.getOptions().containsKey("schema_change_event")) {
+                log.debug("Skipping schema change event row: {}", element.getOptions().keySet());
+                return;
+            }
+        }
+
         if (!submitted) {
             submitted = true;
             runnable.forEach(executorService::submit);