[Pipe] Add an end-to-end full-FLUSH completion barrier and readiness metric

[Caideyipi]

Summary

This PR adds an end-to-end full-FLUSH completion protocol for supported DataNode Pipe tasks and exposes its result through:

pipe_datanode_completion_ready{name,creation_time}

Although the result is exposed as a metric, the implementation spans the StorageEngine and the complete Pipe data path. The existing pipe_datanode_remaining_event_count == 0 does not cover data that is still held by a working TsFile processor, and a realtime source's TsFile/log mode describes only its extraction strategy rather than sink acknowledgement. Neither signal is therefore a safe completion oracle on its own.

The new metric deliberately fails closed: transient false negatives are allowed, but unknown or changing state must never produce a false positive.

Operator contract

The metric reports 1 only when all local DataRegion tasks of the Pipe have passed their latest successful full-FLUSH barriers through:

full FLUSH
-> source assignment
-> processing
-> sink queue
-> sink acknowledgement
-> ordered commit

and the existing remaining-event count is zero. Otherwise it reports 0.

The intended test protocol is:

join/drain writers
-> run a successful full FLUSH covering all relevant DataRegions and both seq/unseq processors
-> freshly scrape this metric directly from every relevant sender DataNode
-> accept completion only when every expected series is 1

When Prometheus is used, callers must additionally verify that:

• every sample timestamp is later than the full FLUSH;
• every expected sender target has up == 1; and
• the returned series cardinality equals the expected sender DataNode count.

The gauge values 0 and 1 are represented exactly. Stale samples and missing targets, rather than floating-point precision, are the relevant risks.

Design and review scope

1. Full-FLUSH boundary in the StorageEngine

• A full FLUSH invalidates the previous completion token before closing processors.
• It captures the working and already-closing seq/unseq TsFile processors present at FLUSH entry.
• It waits for overlapping ordinary asynchronous flushes, then closes the captured processors without chasing processors created by later writes.
• A completion barrier is published only after every captured processor closes successfully. A failed FLUSH never publishes a barrier.
• Inserts invalidate the current token, including inserts ignored by a source mode and inserts that race with a full FLUSH.

2. Barrier publication and source assignment

• PipeDataRegionAssigner serializes data publication and barrier publication and binds each barrier to an assigner epoch and data generation.
• Publication failures, dropped events, stale assigners, and invalidated tokens prevent readiness.
• The barrier follows the same ordered Pipe path as the data it covers.

3. Processing, sink acknowledgement, and ordered commit

• Completion barriers are internal ordering events and cannot be swallowed or transformed by a processor.
• They are not coalesced with periodic heartbeat events in the sink queue.
• Completion is recorded only by the current committer after the sink has acknowledged the barrier and ordered commit has run.

4. Fail-closed readiness evaluation

• The operator takes two task-topology observations and two membership/state observations.
• Source, assigner, task-meta, completion-source, and committer incarnations must remain current.
• Pipe/task exceptions, source degradation, publication failures, task replacement, unsupported tasks, lifecycle changes, stale callbacks, and nonzero remaining events all produce 0.

Supported scope

The metric currently supports a running USER Pipe with:

• iotdb-extractor or iotdb-source;
• do-nothing-processor;
• a built-in IoTDB Thrift sink/connector alias; and
• synchronous TsFile loading.

Unsupported configurations report 0.

The metric covers DataRegion/DML data-point completion only. It does not prove SchemaRegion/DDL completion. The existing remaining-event metric keeps its current meaning, and this metric is not added to the Grafana dashboards in this PR.

Key changed areas

• StorageEngine, DataRegion, and TsFileProcessor: successful full-FLUSH boundary and ordinary-flush overlap handling.
• PipeDataRegionAssigner, realtime sources, and listeners: generation-bound invalidation and barrier publication.
• PipeHeartbeatEvent, processor collection, sink queue, and commit management: ordered barrier propagation through sink acknowledgement.
• PipeDataNodeCompletionOperator and PipeDataNodeSinglePipeMetrics: fail-closed per-DataNode readiness calculation and lifecycle integration.

Validation

• 15 focused Pipe completion/barrier tests: 15 passed, 0 failed/skipped.
• 3 focused full-FLUSH/ordinary-flush concurrency tests: 3 passed, 0 failed/skipped.
• DataNode Checkstyle: 0 violations.
• DataNode Spotless check: 0 files need changes.
• English and Chinese-locale reactor test compilation passed during implementation.

---

This PR has:

• been self-reviewed.
  • concurrent read
  • concurrent write
  • concurrent read and write
• added Javadocs for the new completion operator and non-trivial barrier methods.
• added comments explaining concurrency intent and fail-closed behavior.
• added or updated unit tests for the new code paths.