[Improve][Connector-V2][Prometheus] Flush buffered records on checkpoint to close the Spark/Flink timer-flush gap

[surafel58]

Purpose of this pull request

Closes #11816.

PrometheusWriter buffered records and flushed them only on batch_size, on the Zeta engine timer flush (registerFlushAction), and in close(). It did not flush on checkpoint. On Spark and Flink registerFlushAction keeps the sink writer context's no-op default, so buffered records were held until batch_size was reached or the job closed.

This PR overrides prepareCommit() to flush the buffer and return Optional.empty(), matching the sibling FlushSignal sinks (Doris, ClickHouse, Elasticsearch, StarRocks, MongoDB), which all flush their buffer in prepareCommit(). This bounds the buffered window to one checkpoint interval on every engine, without adding a connector-owned thread or any engine-level change. flush() throws on failure, so a failed checkpoint flush fails the checkpoint instead of silently dropping the batch.

This follows up the #11778 review, where the Spark/Flink gap was flagged and documented as a limitation. The connector-level prepareCommit path used here is the same one the five sibling sinks already use.

Does this PR introduce any user-facing change?

Yes, a behavior improvement. Before, on Spark and Flink, buffered samples were sent only when batch_size was reached or the writer closed, so a low-throughput streaming job could hold points in memory until it stopped. After, the buffer is also flushed on each checkpoint, so buffered samples are bounded by the checkpoint interval on all engines. The batch_size trigger, the Zeta timer flush, and the final flush on close are unchanged.

The Prometheus sink docs and the incompatible-changes.md note (EN and ZH) are updated so the Spark/Flink limitation reads as checkpoint-bounded rather than no periodic flush at all.

How was this patch tested?

Added a unit test shouldFlushOnPrepareCommitWhenEngineNeverInvokesFlushAction in PrometheusWriterTest: with the engine never invoking the registered flush action (the Spark/Flink path), prepareCommit() flushes the buffered row and returns Optional.empty(). The full connector-prometheus module test suite passes locally (10 tests, 0 failures) on JDK 8.

Check list

• If necessary, please update the documentation to describe the new feature. (Prometheus sink docs updated, EN and ZH)
• If necessary, please update incompatible-changes.md to describe the incompatibility caused by this PR. (updated EN and ZH to reflect checkpoint-bounded behavior)
• New Jar binary package: N/A
• New connector: N/A (this modifies an existing connector, no plugin-mapping / seatunnel-dist / label / plugin_config changes needed)

[surafel58]

cc @DanielLeens this implements the checkpoint-flush follow-up you flagged in the #11778 review; scope is as we discussed on #11816. cc @nzw921rx for connector visibility. Whenever you have a chance to take a look, thanks.

[DanielLeens]

What Problem Does This PR Solve?

PrometheusWriter buffers records in batchList and flushes on three triggers: batch_size, the
Zeta engine's timer-flush signal (registerFlushAction), and close(). On Spark and Flink, the
SinkWriter.Context implementations never override registerFlushAction/getFlushAction — they
inherit the interface's no-op default — so the engine-driven timer flush simply never fires on
those two engines (verified: no Flink or Spark Context implementation in the translation modules
overrides these methods; only Zeta's SinkWriterContext does). Worse, PrometheusWriter inherits
HttpSinkWriter.prepareCommit(), which only flushes HttpSinkWriter's own private batchBuffer
field — a field PrometheusWriter never populates, since it overrides write(SeaTunnelRow) to
build a Point and append it to its own batchList instead. So the inherited prepareCommit()
was already a silent no-op for Prometheus's real buffer before this PR. Net effect: on Spark/Flink,
a low-throughput streaming job could hold buffered points in memory indefinitely, across arbitrarily
many checkpoints, until batch_size was reached or the job stopped — an unbounded window in which a
crash loses the buffered (never-POSTed) points. This PR overrides prepareCommit() to call the
already-existing flush() and bounds that window to one checkpoint interval on every engine.

1. Code Change Review

1.1 Core Logic Analysis

@Override
public Optional<Void> prepareCommit() {
    flush();
    return Optional.empty();
}

Checked this against the actual SinkWriter contract (seatunnel-api/.../sink/SinkWriter.java):
prepareCommit() is documented to run "before snapshotState(checkpointId)", so the flush is
guaranteed to happen before the checkpoint's state is finalized — satisfies the "flush before
barrier is acknowledged" requirement. PrometheusWriter has no snapshotState() override (uses
the default empty-list), which is consistent: there is no persisted state to keep in sync, only the
in-flight HTTP push, so a plain "flush now" in prepareCommit() is the right shape.

I verified the invocation path on all three engines, not just trusted the PR description:

• Zeta: SinkFlowLifeCycle.processCheckpointBarrier() calls writer.prepareCommit(barrier.getId())
  inline, from the same single task record-processing loop that also dispatches FlushSignals via
  processSignal(). Both run on the same thread, sequentially — the doc comment's claim of "no
  concurrency between the timer flush and checkpoint... paths" is accurate, confirmed from source,
  not just asserted.
• Flink: FlinkSinkWriter.prepareCommit(boolean) calls sinkWriter.prepareCommit(checkpointId)
  from Flink's own checkpoint lifecycle hook, on the writer's task thread.
• Spark: both SparkDataWriter (2.4) and SeaTunnelSparkDataWriter (3.3) call
  sinkWriter.prepareCommit(epochId) from the DataWriter commit path, again single-threaded per
  writer instance.

So the fix genuinely closes the gap on the two engines named in the PR title, and the existing
synchronized (batchList) in flush() remains defensive-but-correct rather than papering over a
real race.

1.2 Compatibility Impact

This is a behavior-only follow-up to the already-merged, already-documented incompatible change
from #11778 (removal of the connector-owned flush_interval background thread) — it narrows a
previously-documented limitation rather than introducing a new incompatibility. batch_size, the
Zeta timer flush, and the final close() flush are all unchanged; this purely adds one more flush
trigger. EN/ZH docs and EN/ZH incompatible-changes.md are all updated consistently and accurately
reflect the new behavior (I compared the doc wording against the actual code path above and it
matches).

1.3 Performance / Side-Effect Analysis

At-least-once / duplicate-on-retry: if flush() throws inside prepareCommit() (non-204 response
or HTTP failure), the exception fails the checkpoint; since there's no partial state to restore, a
restart replays from the last successful checkpoint and re-sends the same points. Prometheus
remote-write is effectively an upsert keyed by (labels, timestamp), so replaying identical
(timestamp, value) samples is idempotent in effect — this is the same at-least-once characteristic
the sink already had via its batch_size/close() flush paths; this PR doesn't change or worsen
that contract, it only shrinks the exposure window.

One side effect worth flagging explicitly (not called out in the PR/doc text): on Zeta, this adds
an additional flush trigger on top of the existing sink.flush.interval timer. If the job's
checkpoint interval is shorter than sink.flush.interval, Zeta will now flush more often (and in
smaller batches) than before this PR — this is a minor, expected, generally-beneficial side effect
(more consistent output, no behavior change to worry about), but it is a real change in Zeta's
request cadence, not just a Spark/Flink-only change as the PR title suggests. Worth a one-line doc
mention for completeness. See Issue 1.

1.4 Error Handling and Logging

flush()'s existing behavior — throw a PrometheusConnectorException on a non-204 response or
any other failure, propagating instead of silently dropping the batch — is unchanged and is now
also exercised from prepareCommit(). This is exactly right: a failed checkpoint-time flush must
fail the checkpoint, not silently succeed with data unsent. No new logging added or needed here.

2. Code Quality Assessment

2.1 Coding Standards

Clean, minimal diff; the added Javadoc-style inline comment on prepareCommit() clearly explains
why the override is needed and references the sibling connectors for precedent — good practice.
No @DisplayName usage in the test, no single-line Javadoc, consistent with repo conventions.

2.2 Test Coverage and Test Stability

shouldFlushOnPrepareCommitWhenEngineNeverInvokesFlushAction mirrors the existing
shouldFlushOnCloseWhenEngineNeverInvokesFlushAction test precisely (same mock setup, same
"simulate Spark/Flink by never invoking the registered flush action" framing), asserting no POST
before prepareCommit() and exactly one POST after, plus the Optional.empty() return value. This
is a solid, minimal, correctly-targeted test that would fail before this change and pass after it —
genuine regression coverage, not decorative. It's purely additive; no existing test's assertions or
strictness were touched.

2.3 Documentation Updates

EN/ZH sink doc and EN/ZH incompatible-changes.md are all updated, and the wording change ("no
periodic timer flush at all" → "no sub-checkpoint timer flush") accurately reflects the new
checkpoint-bounded behavior. Good documentation discipline, matches this project's requirement that
user-facing behavior changes update both docs trees.

3. Architectural Soundness

3.1 Elegance of the Solution

Reuses the existing flush() method exactly as the sibling FlushSignal sinks (Doris, ClickHouse,
Elasticsearch, StarRocks, MongoDB — verified DorisSinkWriter follows the identical
registerFlushAction + prepareCommit()-flushes pattern) already do. No new connector-owned
thread, no engine-level change — the smallest possible fix that closes the actual gap.

3.2 Maintainability

Trivial to follow; the inline comment explains the "why" (engine differences) rather than restating
the "what". Good.

3.3 Extensibility

Following the same established registerFlushAction + prepareCommit()-flush pattern used by five
other connectors makes this consistent and predictable for anyone extending or reviewing sink
writers in this codebase going forward.

3.4 Historical-Version Compatibility

No config option removed or renamed by this PR itself (that already happened in #11778 and is
already documented); no default value changed; no protocol/serialization change. Purely closes a
behavioral gap on two engines. Safe for upgrade.

4. Issue Summary

Number	Issue	Location	Severity
Issue 1	On Zeta, this adds an additional flush trigger on top of the existing sink.flush.interval timer — if the checkpoint interval is shorter than sink.flush.interval, Zeta now flushes more frequently (smaller batches, more HTTP requests) than before this PR. The PR title/description frame this as closing "the Spark/Flink timer-flush gap", but the Zeta-side cadence change isn't mentioned anywhere in the docs. Not a defect, but worth a one-line note so operators tuning sink.flush.interval vs. checkpoint interval on Zeta aren't surprised by the extra flush trigger.	seatunnel-connectors-v2/connector-prometheus/src/main/java/.../sink/PrometheusWriter.java:117-127; docs/en/connectors/sink/Prometheus.md	Low

Raised by another reviewer: No (no prior reviews/comments exist on this PR besides the author's own
cc-mention to request review; no prior technical feedback to incorporate or dedupe against).

5. Merge Recommendation

Conclusion: Ready to merge

1. Blockers — must be fixed
   • None.
2. Recommended fixes — non-blocking
   • Issue 1: a short doc note that this also changes Zeta's effective flush cadence when the
     checkpoint interval is shorter than sink.flush.interval, so it isn't read as Spark/Flink-only.

Overall assessment: this is a correct, narrowly-scoped, well-tested fix for a genuine
checkpoint-consistency gap. I independently verified the prepareCommit() call path on Zeta, Flink,
and Spark from source rather than taking the PR description's claims at face value, and they all
check out — the flush really does happen before the checkpoint barrier is acknowledged on every
engine, and the "no concurrency between flush and checkpoint" claim is accurate for Zeta's
single-threaded task model. Docs are updated consistently in both languages, and the new test
genuinely exercises the fixed behavior. Good, disciplined follow-up to the earlier #11778 review
feedback — thanks for closing the loop on this.

[DanielLeens]

Approving — the full review above found zero blocking issues. The one note there (Zeta's flush cadence also changes slightly, not just Spark/Flink's) is a non-blocking documentation suggestion.

[surafel58]

Thanks for the thorough review and the approval, and for verifying the prepareCommit path on all three engines. Good catch on Issue 1. Make sense that when the checkpoint interval is shorter than sink.flush.interval, prepareCommit adds an extra flush trigger on Zeta too, not only on Spark and Flink. I will add a one line note to the Prometheus docs (EN and ZH) to make that cadence change explicit and push shortly.

[surafel58]

Done in 4f22513. Added a note to the Prometheus sink docs (EN and ZH) that the checkpoint flush runs on all engines, so on Zeta the buffer is flushed by both sink.flush.interval and each checkpoint, and if the checkpoint interval is shorter than sink.flush.interval the flushes are more frequent. Thanks again for the review.

[nzw921rx]

Thanks for your contribution. My understanding is that the semantics of prepareCommit() here are specifically intended to support 2PC, working together with the engine's checkpoint mechanism to provide exactly-once semantics. With that in mind, I think handling flush() here would be inconsistent with the intended API semantics, so I would prefer that we not support flush() in prepareCommit().