surafel58 opened a new pull request, #11778:
URL: https://github.com/apache/seatunnel/pull/11778
### Purpose of this pull request
Closes #11773.
`PrometheusWriter` used to create and manage its own
`ScheduledExecutorService` to
periodically flush buffered metrics based on the connector-level
`flush_interval`
option. This meant the connector owned an extra background thread whose
lifecycle
could run concurrently with the normal Sink write / checkpoint / close paths.
Zeta already provides engine-level timer flush through `FlushSignal` and
`SinkWriter.Context#registerFlushAction`. This PR migrates the Prometheus
Sink to
that mechanism:
- `PrometheusSink#createWriter()` now passes `SinkWriter.Context` to
`PrometheusWriter`.
- The writer opts in with `context.registerFlushAction(this::flush)`.
- The connector-owned `ScheduledExecutorService`, `ScheduledFuture`, and
`flushSchedule()` (plus their startup/shutdown) are removed.
- The `batch_size` trigger and the final flush in `close()` are preserved.
- A failed flush is now propagated to the engine instead of being logged and
the
batch silently dropped, so a flush that did not succeed is no longer
treated as a
successful flush.
- The connector-level `flush_interval` option is deprecated (kept for
backward
compatibility, but ignored). Users should set the engine-level
`sink.flush.interval`
in the job `env` block instead. If `flush_interval` is still configured,
the writer
logs a one-time warning pointing to the new option.
The engine invokes the registered action on the normal Sink input-processing
path,
so there is no connector-owned scheduler thread and no concurrency with the
Sink
lifecycle. As documented, engine-level timer flush is currently supported
only by Zeta.
### Does this PR introduce _any_ user-facing change?
Yes.
- `flush_interval` (connector option) is deprecated and no longer starts a
background
flush thread. Existing jobs that set it keep validating (the option is
still
accepted) but the value is ignored; a warning is logged. To keep
timer-based
flushing, set `sink.flush.interval` in the job `env` block (Zeta only).
- The English and Chinese Prometheus Sink docs are updated with a new "Timer
Flush"
section and the deprecation note, and the streaming example now uses
`sink.flush.interval`.
### How was this patch tested?
Added `PrometheusWriterTest` (unit test), which verifies:
1. the writer registers a flush action via
`SinkWriter.Context#registerFlushAction`,
and the buffered records are sent only when that action is invoked (the
flush
signal), not on every write;
2. a failed flush (non-`204` response) propagates a
`PrometheusConnectorException`
instead of being swallowed.
Ran locally with JDK 8:
```
./mvnw -pl seatunnel-connectors-v2/connector-prometheus -am install
-DskipTests
./mvnw -pl seatunnel-connectors-v2/connector-prometheus test
-Dtest=PrometheusWriterTest
=> Tests run: 2, Failures: 0, Errors: 0, Skipped: 0
```
`spotless:check` passes for the module.
Note on E2E: the engine already covers the `Source timer -> FlushSignal ->
Sink ->
writer.flush()` path generically in `TimerFlushIT`
(`seatunnel-engine-e2e/connector-seatunnel-e2e-base`), and this PR adds a
connector
unit test for the Prometheus registration/propagation behavior. I have not
added a
Prometheus-specific Zeta E2E in this PR because it needs a Docker-based
Prometheus
mock endpoint that I could not run reliably in my environment. I am happy to
add a
`PrometheusTimerFlushIT` (modeled on the existing `PrometheusIT` container
setup and
the other connectors' `*TimerFlushIT`) as a follow-up or in this PR if the
reviewers
prefer.
### Check list
* [x] If necessary, please update the documentation to describe the new
feature. (EN + ZH Prometheus sink docs updated.)
* [ ] If any new Jar binary package adding in your PR, please add License
Notice — N/A, no new dependencies.
* [ ] Update `incompatible-changes.md` — the `flush_interval` option is
deprecated but still accepted, so no hard incompatibility; happy to add a note
if reviewers want.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
