This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-11827-f9b1b330f19687295970e586bac5acff5e323bb6 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit e46a7106c4cf79853d83e87948a58c1f9708cacf Author: surafel <[email protected]> AuthorDate: Wed Aug 26 06:11:38 2026 +0000 [Improve][Connector-V2][Prometheus] Flush buffered records on checkpoint to close the Spark/Flink timer-flush gap (#11827) --- docs/en/connectors/sink/Prometheus.md | 32 ++++++++++++++--- .../introduction/concepts/incompatible-changes.md | 4 +-- docs/zh/connectors/sink/Prometheus.md | 11 +++++- .../introduction/concepts/incompatible-changes.md | 4 +-- .../prometheus/sink/PrometheusWriter.java | 21 +++++++++++ .../prometheus/sink/PrometheusWriterTest.java | 41 ++++++++++++++++++++++ 6 files changed, 104 insertions(+), 9 deletions(-) diff --git a/docs/en/connectors/sink/Prometheus.md b/docs/en/connectors/sink/Prometheus.md index f8cf6c77ed..0980f63203 100644 --- a/docs/en/connectors/sink/Prometheus.md +++ b/docs/en/connectors/sink/Prometheus.md @@ -98,11 +98,35 @@ connector-owned background thread and no concurrency between the timer flush and checkpoint, or close paths. A flush that fails is propagated to the engine instead of being silently dropped. -> On Spark and Flink there is no periodic timer flush at all: `sink.flush.interval` is a Zeta engine +> On Spark and Flink there is no sub-checkpoint timer flush: `sink.flush.interval` is a Zeta engine > primitive, and the Spark/Flink sink writer context does not implement it. On > those engines the -> buffer is flushed only when it reaches `batch_size` and when the writer is closed. It is **not** -> flushed on checkpoint (`PrometheusWriter` does not override `prepareCommit()`). For a low-throughput -> streaming job on Spark or Flink, tune `batch_size` accordingly. +> buffer is flushed when it reaches `batch_size`, on checkpoint (`PrometheusWriter` flushes in +> `prepareCommit()`), and when the writer is closed. Buffered samples are therefore bounded by the +> checkpoint interval rather than held until `batch_size` or close. For lower latency between +> checkpoints on Spark or Flink, tune `batch_size` accordingly. + +The checkpoint flush runs on all engines, including Zeta. So on Zeta the buffer is flushed by both +`sink.flush.interval` and each checkpoint: if the checkpoint interval is shorter than +`sink.flush.interval`, flushes happen more often (in smaller batches) than the timer alone. This is +expected; tune `sink.flush.interval` and the checkpoint interval together if request cadence matters. + +### Checkpoint Flush and Failure Handling + +The checkpoint flush is a single remote-write request, and a failed flush fails the checkpoint rather +than dropping the batch. Two consequences are worth knowing: + +- **Transient failures fail the checkpoint.** A network blip, a receiver restart, or a `5xx` response + fails the current checkpoint. On Flink the default `tolerableCheckpointFailureNumber` is `0`, so a + single failure restarts the job; on Spark and Flink you may want to raise the engine's tolerable + checkpoint failure setting for a low-throughput job. A bounded retry with backoff inside the flush + is tracked as a follow-up in [#11911](https://github.com/apache/seatunnel/issues/11911). +- **Replay safety depends on the receiver.** After a failed checkpoint the job restarts and the source + replays from the last successful checkpoint, so the buffered samples are re-sent. This is safe only + when the remote-write receiver accepts an exact duplicate (same labels, timestamp, and value). A + receiver that rejects a same-timestamp sample with a different value, or an out-of-order sample + (Prometheus TSDB, and receivers such as Cortex, Mimir, and Thanos, return `400` for these), can fail + the replayed flush and keep failing the checkpoint. Enable the receiver's out-of-order window, or + ensure replays are exact duplicates, if this matters for your deployment. ## Example diff --git a/docs/en/introduction/concepts/incompatible-changes.md b/docs/en/introduction/concepts/incompatible-changes.md index 3f04c6280c..41437be0be 100644 --- a/docs/en/introduction/concepts/incompatible-changes.md +++ b/docs/en/introduction/concepts/incompatible-changes.md @@ -151,9 +151,9 @@ You need to check this document before you upgrade to related version. - **Affected component**: `seatunnel-connectors-v2/connector-prometheus` - **Description**: The Prometheus Sink no longer starts its own background flush thread. The connector-level `flush_interval` option has been removed. Timer-based flushing is now driven by the engine through `sink.flush.interval` in the job `env` block, which is **supported only by the Zeta engine**. - **Impact**: - - **Spark and Flink lose periodic timer-based flushing.** The removed `flush_interval` scheduler was a plain connector-owned thread that ran on all engines. Its replacement, `sink.flush.interval`, is a Zeta engine primitive; the Spark and Flink sink writer contexts do not implement it, so there is no periodic flush on those engines. On Spark and Flink the buffer is now flushed only when it reaches `batch_size` and when the writer is closed (it is not flushed on checkpoint). A low-thr [...] + - **Spark and Flink lose sub-checkpoint timer-based flushing.** The removed `flush_interval` scheduler was a plain connector-owned thread that ran on all engines. Its replacement, `sink.flush.interval`, is a Zeta engine primitive; the Spark and Flink sink writer contexts do not implement it, so there is no periodic timer flush on those engines. On Spark and Flink the buffer is flushed when it reaches `batch_size`, on checkpoint (the sink flushes in `prepareCommit()`), and when the wr [...] - A leftover `flush_interval` key in the `Prometheus` sink block is rejected only when the config is validated with `--check` / `--dry-run=static` / `--dry-run=connect` (which run `validateUnknownKeys`). A directly submitted job silently ignores the stray key; the connector logs a warning once per sink writer at startup instead (so a job with parallelism N, multiple tables, or replicas logs it multiple times). - - **Migration Guide**: Remove `flush_interval` from the `Prometheus` sink block. To keep timer-based flushing on Zeta, set `sink.flush.interval` (milliseconds) in the job `env` block. On Spark and Flink, rely on `batch_size`. The `batch_size` trigger and the final flush on writer close are unchanged on all engines. + - **Migration Guide**: Remove `flush_interval` from the `Prometheus` sink block. To keep timer-based flushing on Zeta, set `sink.flush.interval` (milliseconds) in the job `env` block. On Spark and Flink, buffered points are flushed on each checkpoint; tune `batch_size` for lower latency between checkpoints. The `batch_size` trigger and the final flush on writer close are unchanged on all engines. - **Breaking Change: File connectors reject `DOCTYPE` declarations in XML input (XXE hardening)** - **Affected component**: `seatunnel-connectors-v2/connector-file/connector-file-base` (`XmlReadStrategy`), and every file source built on it: LocalFile, HdfsFile, S3File, OssFile, OssJindoFile, CosFile, FtpFile, SftpFile (`file_format_type = xml`) diff --git a/docs/zh/connectors/sink/Prometheus.md b/docs/zh/connectors/sink/Prometheus.md index 558f0ca0db..b528d9cf7b 100644 --- a/docs/zh/connectors/sink/Prometheus.md +++ b/docs/zh/connectors/sink/Prometheus.md @@ -88,7 +88,16 @@ env { 引擎会在正常的 Sink 数据处理线程上触发刷新，因此不需要连接器自己维护后台线程，也不会和写入、检查点、关闭等流程产生并发。刷新失败会被抛给引擎，而不会被静默丢弃。 -> 在 Spark 和 Flink 上完全没有周期性定时刷新：`sink.flush.interval` 是 Zeta 引擎的能力，Spark/Flink 的 Sink 写入器上下文并未实现它。在这两个引擎上，缓存只会在达到 `batch_size` 以及写入器关闭时被刷新，**不会**在检查点时刷新（`PrometheusWriter` 未重写 `prepareCommit()`）。对于低吞吐的流式作业，请相应调整 `batch_size`。 +> 在 Spark 和 Flink 上没有检查点之间的定时刷新：`sink.flush.interval` 是 Zeta 引擎的能力，Spark/Flink 的 Sink 写入器上下文并未实现它。在这两个引擎上，缓存会在达到 `batch_size`、检查点时（`PrometheusWriter` 在 `prepareCommit()` 中刷新）以及写入器关闭时被刷新。因此缓存的采样点最多保留一个检查点间隔，而不会一直保存到 `batch_size` 或关闭。如需降低 Spark 或 Flink 上检查点之间的延迟，请相应调整 `batch_size`。 + +检查点刷新在所有引擎上都会执行，包括 Zeta。因此在 Zeta 上，缓存会同时由 `sink.flush.interval` 和每个检查点触发刷新：如果检查点间隔短于 `sink.flush.interval`，刷新会比仅靠定时器时更频繁（每批更小）。这是预期行为；如果关注请求频率，请同时调整 `sink.flush.interval` 和检查点间隔。 + +### 检查点刷新与失败处理 + +检查点刷新是一次 remote-write 请求，刷新失败会让检查点失败，而不会丢弃这批数据。有两点需要了解： + +- **瞬时失败会导致检查点失败。** 网络抖动、接收端重启或 `5xx` 响应都会让当前检查点失败。Flink 的 `tolerableCheckpointFailureNumber` 默认是 `0`，因此一次失败就会重启作业；在 Spark 和 Flink 上，对于低吞吐作业你可能需要调高引擎的可容忍检查点失败次数。刷新内部的有界重试（带退避）作为后续项跟踪在 [#11911](https://github.com/apache/seatunnel/issues/11911)。 +- **重放的安全性取决于接收端。** 检查点失败后作业会重启，Source 从上一次成功的检查点重放，因此缓存的采样点会被重新发送。只有当 remote-write 接收端接受完全相同的重复样本（相同的 labels、timestamp 和 value）时，这才是安全的。如果接收端拒绝相同 timestamp 但 value 不同的样本，或拒绝乱序样本（Prometheus TSDB，以及 Cortex、Mimir、Thanos 等接收端对这些情况会返回 `400`），重放的刷新就会失败并持续让检查点失败。如果这对你的部署很重要，请启用接收端的乱序窗口，或确保重放的是完全相同的重复样本。 ## 示例 diff --git a/docs/zh/introduction/concepts/incompatible-changes.md b/docs/zh/introduction/concepts/incompatible-changes.md index 130eed2408..7616859aef 100644 --- a/docs/zh/introduction/concepts/incompatible-changes.md +++ b/docs/zh/introduction/concepts/incompatible-changes.md @@ -141,9 +141,9 @@ - **受影响组件**：`seatunnel-connectors-v2/connector-prometheus` - **变更说明**：Prometheus Sink 不再启动自己的后台刷新线程，连接器级的 `flush_interval` 选项已被移除。定时刷新改为由引擎通过作业 `env` 中的 `sink.flush.interval` 驱动，**仅 Zeta 引擎支持**。 - **影响**： - - **Spark 和 Flink 会失去周期性定时刷新。** 被移除的 `flush_interval` 调度器是连接器自己的线程，在所有引擎上都能工作；其替代者 `sink.flush.interval` 是 Zeta 引擎的能力，Spark 和 Flink 的 Sink 写入器上下文并未实现它，因此这两个引擎上没有周期性刷新。在 Spark 和 Flink 上，缓存现在只会在达到 `batch_size` 以及写入器关闭时被刷新（不会在检查点时刷新）。因此低吞吐的流式作业可能会把缓存的采样点一直保存在内存中直到作业停止；请相应调整 `batch_size`。 + - **Spark 和 Flink 会失去检查点之间的定时刷新。** 被移除的 `flush_interval` 调度器是连接器自己的线程，在所有引擎上都能工作；其替代者 `sink.flush.interval` 是 Zeta 引擎的能力，Spark 和 Flink 的 Sink 写入器上下文并未实现它，因此这两个引擎上没有周期性定时刷新。在 Spark 和 Flink 上，缓存会在达到 `batch_size`、检查点时（Sink 在 `prepareCommit()` 中刷新）以及写入器关闭时被刷新。因此缓存的采样点最多保留一个检查点间隔，而不会一直保存到作业停止；如需降低检查点之间的延迟，请相应调整 `batch_size`。 - 只有在使用 `--check` / `--dry-run=static` / `--dry-run=connect` 校验配置时（会执行 `validateUnknownKeys`），`Prometheus` sink 中残留的 `flush_interval` 键才会被拒绝。直接提交的作业会静默忽略该残留键；连接器会在每个 Sink 写入器启动时各打印一次告警作为替代提示（因此并行度为 N、多表或多副本的作业会多次打印）。 - - **迁移指南**：从 `Prometheus` sink 中移除 `flush_interval`。如需在 Zeta 上继续使用定时刷新，请在作业 `env` 中设置 `sink.flush.interval`（毫秒）。在 Spark 和 Flink 上请依赖 `batch_size`。`batch_size` 触发和写入器关闭时的最后一次刷新在所有引擎上保持不变。 + - **迁移指南**：从 `Prometheus` sink 中移除 `flush_interval`。如需在 Zeta 上继续使用定时刷新，请在作业 `env` 中设置 `sink.flush.interval`（毫秒）。在 Spark 和 Flink 上，缓存会在每个检查点被刷新；如需降低检查点之间的延迟，请调整 `batch_size`。`batch_size` 触发和写入器关闭时的最后一次刷新在所有引擎上保持不变。 - **破坏性变更：File 连接器拒绝 XML 输入中的 `DOCTYPE` 声明（XXE 加固）** - **影响范围**：`seatunnel-connectors-v2/connector-file/connector-file-base`（`XmlReadStrategy`），以及所有基于该模块构建的 File Source：LocalFile、HdfsFile、S3File、OssFile、OssJindoFile、CosFile、FtpFile、SftpFile（`file_format_type = xml`） diff --git a/seatunnel-connectors-v2/connector-prometheus/src/main/java/org/apache/seatunnel/connectors/seatunnel/prometheus/sink/PrometheusWriter.java b/seatunnel-connectors-v2/connector-prometheus/src/main/java/org/apache/seatunnel/connectors/seatunnel/prometheus/sink/PrometheusWriter.java index f1de9b2c91..a1eec20251 100644 --- a/seatunnel-connectors-v2/connector-prometheus/src/main/java/org/apache/seatunnel/connectors/seatunnel/prometheus/sink/PrometheusWriter.java +++ b/seatunnel-connectors-v2/connector-prometheus/src/main/java/org/apache/seatunnel/connectors/seatunnel/prometheus/sink/PrometheusWriter.java @@ -43,6 +43,7 @@ import java.io.IOException; import java.util.ArrayList; import java.util.List; import java.util.Map; +import java.util.Optional; @Slf4j public class PrometheusWriter extends HttpSinkWriter { @@ -113,6 +114,26 @@ public class PrometheusWriter extends HttpSinkWriter { } } + @Override + public Optional<Void> prepareCommit() { + // Flush buffered records on checkpoint. On Spark and Flink the engine-level timer flush + // (registerFlushAction) keeps the Context's no-op default, so without this the buffer would + // only be sent on batch_size and on close(). Flushing on checkpoint bounds the buffered + // window to one checkpoint interval on every engine, matching the sibling FlushSignal sinks + // (Doris, ClickHouse, Elasticsearch, StarRocks, MongoDB), which flush their buffer here for + // the non-2PC case and return Optional.empty(). Prometheus remote-write is not + // transactional, so there is no commit info to return. + // + // flush() throws on failure, so a failed checkpoint flush fails the checkpoint instead of + // silently dropping the batch. On restart the source replays from the last successful + // checkpoint and re-sends the buffered samples. Whether that replay is harmless depends on + // the receiver: one that treats a repeated (labels, timestamp) sample as an idempotent + // upsert absorbs it, but one that rejects duplicate or out-of-order samples may fail the + // replayed flush, so the delivery guarantee here is at-least-once, not exactly-once. + flush(); + return Optional.empty(); + } + private void flush() { synchronized (batchList) { if (batchList.isEmpty()) { diff --git a/seatunnel-connectors-v2/connector-prometheus/src/test/java/org/apache/seatunnel/connectors/seatunnel/prometheus/sink/PrometheusWriterTest.java b/seatunnel-connectors-v2/connector-prometheus/src/test/java/org/apache/seatunnel/connectors/seatunnel/prometheus/sink/PrometheusWriterTest.java index ac769bf96f..26a7a3aa2b 100644 --- a/seatunnel-connectors-v2/connector-prometheus/src/test/java/org/apache/seatunnel/connectors/seatunnel/prometheus/sink/PrometheusWriterTest.java +++ b/seatunnel-connectors-v2/connector-prometheus/src/test/java/org/apache/seatunnel/connectors/seatunnel/prometheus/sink/PrometheusWriterTest.java @@ -39,6 +39,7 @@ import org.mockito.MockedConstruction; import java.io.IOException; import java.util.HashMap; import java.util.Map; +import java.util.Optional; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; @@ -148,6 +149,46 @@ class PrometheusWriterTest { } } + /** + * On Spark and Flink the engine never invokes the flush action, but prepareCommit() runs on + * every checkpoint. The buffered records must be delivered at checkpoint time instead of + * waiting for batch_size or close(), which bounds the buffered window to one checkpoint + * interval on those engines and matches the sibling FlushSignal sinks. + */ + @Test + void shouldFlushOnPrepareCommitWhenEngineNeverInvokesFlushAction() throws Exception { + SinkWriter.Context context = mock(SinkWriter.Context.class); + HttpResponse ok = new HttpResponse(HttpStatus.SC_NO_CONTENT); + + try (MockedConstruction<HttpClientProvider> ignored = + mockConstruction( + HttpClientProvider.class, + (mockClient, ctx) -> + when(mockClient.doPost( + anyString(), any(), any(ByteArrayEntity.class))) + .thenReturn(ok))) { + + PrometheusWriter writer = createWriter(context); + writer.write(newPoint()); + + // Simulate Spark/Flink: the registered flush action is never invoked by the engine. + verify(writer.httpClient, never()) + .doPost(anyString(), any(), any(ByteArrayEntity.class)); + + Assertions.assertEquals(Optional.empty(), writer.prepareCommit()); + + // prepareCommit() must flush the buffered row so it reaches Prometheus at checkpoint. + verify(writer.httpClient, times(1)) + .doPost(anyString(), any(), any(ByteArrayEntity.class)); + + // The buffer must be cleared after the flush, so a later checkpoint with no new rows + // does not re-deliver the same row: a second prepareCommit() sends nothing more. + Assertions.assertEquals(Optional.empty(), writer.prepareCommit()); + verify(writer.httpClient, times(1)) + .doPost(anyString(), any(), any(ByteArrayEntity.class)); + } + } + /** * When the final flush in close() fails and closing the HTTP client also throws, the meaningful * flush failure must be the exception that surfaces, with the client-close error suppressed,
