[Feature][Zeta] STIP-23 Phase 1: Add engine timer flush core flow and FlushSignal handling

docs/en/introduction/configuration/JobEnvConfig.md

@@ -55,6 +55,12 @@ This parameter is used to specify the location of the savemode when the job is e
 The default value is `CLUSTER`, which means that the savemode is executed on the cluster. If you want to execute the savemode on the client,
 you can set it to `CLIENT`. Please use `CLUSTER` mode as much as possible, because when there are no problems with `CLUSTER` mode, we will remove `CLIENT` mode.
 
+### sink.flush.interval
+
+Interval (ms) at which the engine injects a `FlushSignal` into the pipeline to drive a flush at the Sink. `0` or unset (default) means disabled. Only works in the Zeta engine.
+
+Values below 100ms are not recommended — excessive signals consume pipeline queue capacity, crowding out normal data records, and trigger empty flushes when no data has been buffered yet, increasing Sink I/O overhead.
+
 ## Flink Engine Parameter
 
 Here are some SeaTunnel parameter names corresponding to the names in Flink, not all of them. Please refer to the official [Flink Documentation](https://flink.apache.org/).

docs/zh/introduction/configuration/JobEnvConfig.md

@@ -56,6 +56,12 @@
 当值为`CLIENT`时，SaveMode操作在作业提交的过程中执行，使用shell脚本提交作业时，该过程在提交作业的shell进程中执行。使用rest api提交作业时，该过程在http请求的处理线程中执行。
 请尽量使用`CLUSTER`模式，因为当`CLUSTER`模式没有问题时，我们将删除`CLIENT`模式。
 
+### sink.flush.interval
+
+定时向数据流中注入 `FlushSignal` 的间隔（毫秒），驱动 Sink 刷写缓冲数据。设置为 `0` 或不配置（默认）时不生效。仅适用于 Zeta 引擎。
+
+建议不低于 100ms。过于频繁会产生大量无效空刷信号占用数据流队列容量，挤压正常数据记录的传输空间，并在缓冲区尚无数据时触发无意义的写出，增加 Sink I/O 开销。
+
 ## Flink 引擎参数
 
 这里列出了一些与 Flink 中名称相对应的 SeaTunnel 参数名称，并非全部，更多内容请参考官方 [Flink Documentation](https://flink.apache.org/) for more.

seatunnel-api/src/main/java/org/apache/seatunnel/api/options/EnvCommonOptions.java

@@ -85,6 +85,14 @@ public class EnvCommonOptions {
                     .noDefaultValue()
                     .withDescription("The timeout (in milliseconds) for a checkpoint.");
 
+    public static Option<Long> SINK_FLUSH_INTERVAL =
+            Options.key("sink.flush.interval")
+                    .longType()
+                    .defaultValue(0L)
+                    .withDescription(
+                            "Interval (ms) at which the engine injects a FlushSignal into the pipeline to "
+                                    + "drive a flush at the Sink. 0 means disabled. Values below 100ms will log a WARN.");
+
     public static Option<Integer> CHECKPOINT_MIN_PAUSE =
             Options.key("min-pause")
                     .intType()

seatunnel-api/src/main/java/org/apache/seatunnel/api/options/EnvOptionRule.java

@@ -47,7 +47,8 @@ public OptionRule optionRule() {
                         EnvCommonOptions.READ_LIMIT_BYTES_PER_SECOND,
                         EnvCommonOptions.SAVEMODE_EXECUTE_LOCATION,
                         EnvCommonOptions.CUSTOM_PARAMETERS,
-                        EnvCommonOptions.NODE_TAG_FILTER)
+                        EnvCommonOptions.NODE_TAG_FILTER,
+                        EnvCommonOptions.SINK_FLUSH_INTERVAL)
                 .build();
     }
 }

seatunnel-api/src/main/java/org/apache/seatunnel/api/signal/FlushSignal.java

@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.api.signal;
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+import java.util.Objects;
+
+@Getter
+@RequiredArgsConstructor
+public final class FlushSignal implements Signal {
+
+    private static final long serialVersionUID = 1L;
+
+    private final long jobId;
+    private final long taskId;
+    private final long createdTime;
+
+    public static FlushSignal of(long jobId, long taskId) {
+        return new FlushSignal(jobId, taskId, System.currentTimeMillis());
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!(o instanceof FlushSignal)) {
+            return false;
+        }
+        FlushSignal that = (FlushSignal) o;
+        return jobId == that.jobId && taskId == that.taskId && createdTime == that.createdTime;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(jobId, taskId, createdTime);
+    }
+
+    @Override
+    public String toString() {
+        return "FlushSignal{"
+                + "jobId="
+                + jobId
+                + ", taskId="
+                + taskId
+                + ", createdTime="
+                + createdTime
+                + '}';
+    }
+}

seatunnel-api/src/main/java/org/apache/seatunnel/api/signal/Signal.java

@@ -0,0 +1,52 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.seatunnel.api.signal;
+
+import java.io.Serializable;
+
+/**
+ * Interface for control-plane signals that the engine propagates through the data flow.
+ *
+ * <p>A {@code Signal} is not a data record: it carries no user payload and is intended to instruct
+ * an operator (e.g. a {@link org.apache.seatunnel.api.sink.SinkWriter}) to perform a side-effecting
+ * action such as flushing buffered data.
+ *
+ * <p>Every signal carries three pieces of metadata:
+ *
+ * <ul>
+ *   <li>{@link #getJobId()} — the job that produced the signal.
+ *   <li>{@link #getTaskId()} — the source task that produced the signal, useful for logging and
+ *       per-subtask diagnostics.
+ *   <li>{@link #getCreatedTime()} — the wall-clock time (epoch millis) at which the signal was
+ *       created by the engine.
+ * </ul>
+ *
+ * <p>Concrete signals should be small and immutable. New signal types are added by introducing new
+ * implementations of this interface; the engine routes them by type via {@code instanceof}.
+ */
+public interface Signal extends Serializable {
+
+    /** @return the id of the job that created this signal. */
+    long getJobId();
+
+    /** @return the id of the task that created this signal. */
+    long getTaskId();
+
+    /** @return the wall-clock creation time of this signal, in epoch milliseconds. */
+    long getCreatedTime();
+}

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

@@ -20,6 +20,7 @@
 import org.apache.seatunnel.api.common.metrics.MetricsContext;
 import org.apache.seatunnel.api.event.EventListener;
 import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
+import org.apache.seatunnel.common.utils.function.RunnableWithException;
 
 import java.io.IOException;
 import java.io.Serializable;
@@ -117,5 +118,28 @@ default int getNumberOfParallelSubtasks() {
          * @return
          */
         EventListener getEventListener();
+
+        /**
+         * Register an action to be invoked by the engine when a periodic flush signal arrives.
+         *
+         * <p>This is the opt-in point for engine-level timer flush. A writer that wants to be
+         * flushed on a schedule should call this method during its initialization, typically with a
+         * method reference like {@code context.registerFlushAction(this::flush)}.
+         *
+         * @param action the action to invoke on each flush signal, must not be {@code null}
+         */
+        default void registerFlushAction(RunnableWithException action) {}
+
+        /**
+         * Return the flush action previously registered via {@link
+         * #registerFlushAction(RunnableWithException)}, or {@code null} if the writer has not opted
+         * in to engine-level timer flush.
+         *
+         * <p>Callers must null-check the return value; a {@code null} return means the writer will
+         * silently ignore flush signals.
+         */
+        default RunnableWithException getFlushAction() {
+            return null;
+        }
     }
 }

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

@@ -102,18 +102,22 @@ public SinkWriter<SeaTunnelRow, MultiTableCommitInfo, MultiTableState> createWri
             SinkWriter.Context context) throws IOException {
         Map<SinkIdentifier, SinkWriter<SeaTunnelRow, ?, ?>> writers = new HashMap<>();
         Map<SinkIdentifier, SinkWriter.Context> sinkWritersContext = new HashMap<>();
+        Map<SinkIdentifier, SinkContextProxy> proxyContexts = new HashMap<>();
         for (int i = 0; i < replicaNum; i++) {
             for (TablePath tablePath : sinks.keySet()) {
                 SeaTunnelSink sink = sinks.get(tablePath);
                 int index = context.getIndexOfSubtask() * replicaNum + i;
-                String tableIdentifier = tablePath.toString();
-                writers.put(
-                        SinkIdentifier.of(tableIdentifier, index),
-                        sink.createWriter(new SinkContextProxy(index, replicaNum, context)));
-                sinkWritersContext.put(SinkIdentifier.of(tableIdentifier, index), context);
+                SinkIdentifier id = SinkIdentifier.of(tablePath.toString(), index);
+                SinkContextProxy proxy = new SinkContextProxy(index, replicaNum, context);
+                writers.put(id, sink.createWriter(proxy));
+                proxyContexts.put(id, proxy);
+                sinkWritersContext.put(id, context);
             }
         }
-        return new MultiTableSinkWriter(writers, replicaNum, sinkWritersContext);
+        MultiTableSinkWriter writer =
+                new MultiTableSinkWriter(writers, replicaNum, sinkWritersContext);
+        registerAggregatedFlushIfNeeded(context, writer, proxyContexts);
+        return writer;
     }
 
     /**
@@ -134,12 +138,14 @@ public SinkWriter<SeaTunnelRow, MultiTableCommitInfo, MultiTableState> restoreWr
             SinkWriter.Context context, List<MultiTableState> states) throws IOException {
         Map<SinkIdentifier, SinkWriter<SeaTunnelRow, ?, ?>> writers = new HashMap<>();
         Map<SinkIdentifier, SinkWriter.Context> sinkWritersContext = new HashMap<>();
+        Map<SinkIdentifier, SinkContextProxy> proxyContexts = new HashMap<>();
 
         for (int i = 0; i < replicaNum; i++) {
             for (TablePath tablePath : sinks.keySet()) {
                 SeaTunnelSink sink = sinks.get(tablePath);
                 int index = context.getIndexOfSubtask() * replicaNum + i;
                 SinkIdentifier sinkIdentifier = SinkIdentifier.of(tablePath.toString(), index);
+                SinkContextProxy proxy = new SinkContextProxy(index, replicaNum, context);
                 List<?> state =
                         states.stream()
                                 .map(
@@ -149,19 +155,36 @@ public SinkWriter<SeaTunnelRow, MultiTableCommitInfo, MultiTableState> restoreWr
                                 .flatMap(Collection::stream)
                                 .collect(Collectors.toList());
                 if (state.isEmpty()) {
-                    writers.put(
-                            sinkIdentifier,
-                            sink.createWriter(new SinkContextProxy(index, replicaNum, context)));
+                    writers.put(sinkIdentifier, sink.createWriter(proxy));
                 } else {
-                    writers.put(
-                            sinkIdentifier,
-                            sink.restoreWriter(
-                                    new SinkContextProxy(index, replicaNum, context), state));
+                    writers.put(sinkIdentifier, sink.restoreWriter(proxy, state));
                 }
+                proxyContexts.put(sinkIdentifier, proxy);
                 sinkWritersContext.put(sinkIdentifier, context);
             }
         }
-        return new MultiTableSinkWriter(writers, replicaNum, sinkWritersContext);
+        MultiTableSinkWriter writer =
+                new MultiTableSinkWriter(writers, replicaNum, sinkWritersContext);
+        registerAggregatedFlushIfNeeded(context, writer, proxyContexts);
+        return writer;
+    }
+
+    /**
+     * Registers an aggregated flush action on the parent context if any sub-writer registered a
+     * flush action via its {@link SinkContextProxy}.
+     *
+     * <p>The registered action drains all blocking queues and then calls each sub-writer's flush
+     * action under the corresponding lock, ensuring safe execution from the engine timer thread.
+     */
+    private void registerAggregatedFlushIfNeeded(
+            SinkWriter.Context context,
+            MultiTableSinkWriter writer,
+            Map<SinkIdentifier, SinkContextProxy> proxyContexts) {
+        boolean anyFlush =
+                proxyContexts.values().stream().anyMatch(p -> p.getFlushAction() != null);
+        if (anyFlush) {
+            context.registerFlushAction(() -> writer.aggregatedFlush(proxyContexts));
+        }
     }
 
     @Override

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

@@ -485,6 +485,28 @@ public void close() throws IOException {
         }
     }
 
+    /**
+     * Aggregated flush triggered by the engine timer. Drains all blocking queues first, then calls
+     * each sub-writer's flush action under the corresponding {@link MultiTableWriterRunnable} lock
+     * to prevent concurrent writes.
+     *
+     * @param proxyContexts map from sink identifier to its {@link SinkContextProxy}
+     */
+    void aggregatedFlush(Map<SinkIdentifier, SinkContextProxy> proxyContexts) throws Exception {
+        checkQueueRemain();
+        subSinkErrorCheck();
+        for (int i = 0; i < sinkWritersWithIndex.size(); i++) {
+            synchronized (runnable.get(i)) {
+                for (SinkIdentifier id : sinkWritersWithIndex.get(i).keySet()) {
+                    SinkContextProxy proxy = proxyContexts.get(id);
+                    if (proxy != null && proxy.getFlushAction() != null) {
+                        proxy.getFlushAction().run();
+                    }
+                }
+            }
+        }
+    }
+
     /**
      * Busy-waits until all blocking queues are fully drained.
      *

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

@@ -20,14 +20,16 @@
 import org.apache.seatunnel.api.common.metrics.MetricsContext;
 import org.apache.seatunnel.api.event.EventListener;
 import org.apache.seatunnel.api.sink.SinkWriter;
+import org.apache.seatunnel.common.utils.function.RunnableWithException;
+
+import java.util.Objects;
 
 public class SinkContextProxy implements SinkWriter.Context {
 
     private final int index;
-
     private final int replicaNum;
-
     private final SinkWriter.Context context;
+    private transient volatile RunnableWithException flushAction;
 
     public SinkContextProxy(int index, int replicaNum, SinkWriter.Context context) {
         this.index = index;
@@ -54,4 +56,15 @@ public MetricsContext getMetricsContext() {
     public EventListener getEventListener() {
         return context.getEventListener();
     }
+
+    @Override
+    public void registerFlushAction(RunnableWithException action) {
+        Objects.requireNonNull(action, "flushAction");
+        this.flushAction = action;
+    }
+
+    @Override
+    public RunnableWithException getFlushAction() {
+        return flushAction;
+    }
 }