[Fix][Flink] Supports multiple parallelisms and coordinator state persistence

Author: CloverDew

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

@@ -17,8 +17,6 @@
 
 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;
@@ -32,48 +30,4 @@ 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,7 +21,6 @@
 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;
@@ -176,26 +175,6 @@ 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) {

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

@@ -50,6 +50,14 @@ public void run() {
                 if (row == null) {
                     continue;
                 }
+                // control rows used for schema evolution / coordination
+                // are represented as SeaTunnelRow with zero fields (arity == 0)
+                if (row.getArity() == 0) {
+                    log.debug(
+                            "Skip control SeaTunnelRow with zero arity in MultiTableWriterRunnable: {}",
+                            row);
+                    continue;
+                }
                 SinkWriter<SeaTunnelRow, ?, ?> writer = tableIdWriterMap.get(row.getTableId());
                 if (writer == null) {
                     if (tableIdWriterMap.size() == 1) {