[Pipe] Add reliable DataNode completion metric by Caideyipi · Pull Request #18563 · apache/iotdb

Caideyipi · 2026-09-01T11:29:00Z

Pipe Completion Metric Design

1. Conclusion

This implementation retains the complete end-to-end barrier protocol from Apache IoTDB PR #18280 and adds expected-DataRegion integrity checks on the current branch.

The new metric is:

pipe_datanode_completion_ready{name="<pipeName>",creation_time="<creationTime>"}

The metric has exactly two possible values, 0 and 1:

1: On this sender DataNode, the latest full-FLUSH completion barrier for every DataRegion currently expected by this Pipe has passed through the source, processor, and sink, and has been committed in order after a successful sink ACK. There are no pending non-heartbeat events, and no changes have been detected in task, source, assigner, committer, exception, or degraded state.
0: The operation is incomplete, unsupported, unknown, affected by a race, or failed a health check. The value 0 does not distinguish among these reasons.

The design deliberately allows temporary false negatives but never false positives. The existing remaining_event_count remains a progress metric: a value of 0 is necessary for completion, but it is not sufficient proof that sending has completed.

2. Why `remaining_event_count` cannot simply include "waiting for flush" and "captured TsFile"

A smaller implementation could add the storage engine's processors waiting to close, captured TsFiles, and Pipe queues together, but that would provide only an approximate backlog and could not reliably prove that sending had completed.

2.1 State transitions create counting gaps

The same TsFile goes through:

working/closing processor -> close callback -> assigner -> source queue -> processor queue -> sink/in-flight RPC

If these structures are read and summed separately, an object moving between two structures may temporarily belong to neither of the snapshots that have been read, producing a transient 0. Adding more containers only narrows the window; it cannot eliminate a cross-component snapshot race.

2.2 An empty queue does not mean that the sink has confirmed receipt

After an event is taken from the sink queue and before the destination returns a successful ACK, the queue may already be empty. At that point, remaining_event_count == 0 still does not prove that the destination has received or loaded the data.

2.3 FLUSH and concurrent writes have a boundary race

After a full FLUSH starts, a concurrent insert may create a new working processor that is not included in the capture set for that FLUSH. Sending all older TsFiles therefore does not mean that the current round of writes is complete. Every full FLUSH must establish a token, and a concurrent insert must invalidate that token.

2.4 Lifecycle changes can make old state report completion incorrectly

If a Pipe task, realtime source, assigner, or committer is replaced, state committed by the old instance cannot prove that the new instance is complete. If task initialization fails and a DataRegion is missing, the absence of a local task must not be interpreted as completion.

Therefore, if the goal is only an approximate backlog, a small change to remaining_event_count is sufficient. If the goal is to determine reliably that a Pipe has finished sending, an end-to-end ordered barrier is required. The main purpose of #18280 is to close the false-positive windows described above; retaining only one of its counting points would not be sufficient.

3. Completion barrier flow

stop and join writers
        |
        v
run a full FLUSH covering all relevant DataRegions
        |
        +-- invalidate the old completion token and record a new token
        +-- capture the working and closing TsFileProcessors at FLUSH entry
        +-- wait for ordinary asynchronous flushes already in progress
        +-- close and await the captured processors and their close callbacks
        +-- publish the barrier only if the token was not invalidated by a concurrent insert
        |
        v
assigner (bound to the assigner epoch and data generation)
        |
        v
realtime source -> processor (the barrier cannot be swallowed or rewritten)
        |
        v
sink queue (the barrier is not coalesced with ordinary heartbeat events)
        |
        v
successful sink ACK -> ordered commit -> onCommitted hook
        |
        v
completion operator: double-snapshot validation
        |
        v
pipe_datanode_completion_ready = 1

The key point is that the barrier is ordered after the TsFile events captured and published by this FLUSH. A DataRegion is marked complete only after the sink has successfully processed the barrier and the ordered commit has completed.

4. Fail-closed conditions

The metric returns 0 if any of the following conditions holds:

remaining_event_count still contains a non-heartbeat event;
the Pipe does not exist, is not a RUNNING USER Pipe, or has a Pipe/runtime/task exception;
the source, processor, sink, or TsFile load strategy is unsupported;
the historical source has not been fully consumed, or the realtime source has not started completely;
the actual DataRegion task set differs from the expected set calculated from PipeMeta, the leader, and the local StorageEngine;
a task, source, assigner, or committer instance has been replaced;
the full-FLUSH token was invalidated by a concurrent insert;
the barrier generation is behind the latest data generation;
event publication, reference counting, enqueueing, or delivery fails;
a hybrid source is waiting for TsFile recovery to restore a discarded tablet (degraded);
any change occurs between the two task-topology snapshots or between the two state snapshots;
a runtime exception occurs during metric calculation, or acquiring the task read lock times out.

Lost events and publication failures cannot be automatically proven to have recovered. The metric therefore remains 0; the usual recovery is to fix the problem, restart or rebuild the relevant Pipe tasks, and execute the completion protocol again.

5. Supported scope

Currently, only the following combinations can return 1:

a RUNNING USER Pipe;
source/extractor: iotdb-extractor or iotdb-source;
processor: do-nothing-processor;
sink/connector: a built-in IoTDB Thrift connector/sink, including its sync, async, SSL, and other aliases;
TsFile load strategy: sync;
the source has started and the historical phase has been fully consumed; and
DataRegion/DML transmission has completed.

This metric does not currently prove that SchemaRegion/DDL operations have completed, and it provides no completion guarantee for custom processors, custom sinks, or asynchronous TsFile load strategies. These combinations conservatively return 0.

6. Correct use of the protocol

Stop all writers that may write to the relevant DataRegions and wait for their threads to finish. The current generation is scoped to a DataRegion; even concurrent writes outside the Pipe pattern may conservatively keep the metric at 0.
Run a successful full FLUSH covering every DataRegion that the Pipe may send. A global full FLUSH is recommended. Do not use a partial flush that closes only SEQ or UNSEQ processors as the completion barrier.
After the FLUSH returns successfully, obtain a fresh metric sample on each expected sender DataNode.
Conclude that this round of Pipe DML transmission is complete only when all expected series exist and every value is 1.

In a Prometheus deployment, also confirm that:

the sample timestamp is later than this FLUSH;
every target has up == 1;
the number of series matches the expected number of sender DataNodes; and
missing series, scrape failures, and stale samples are not treated as completion.

creation_time distinguishes different incarnations of a Pipe with the same name. The evaluation must be pinned to the exact name + creation_time for this incarnation; do not aggregate old series by name alone.

7. Relationship to PR #18280 and adaptation on the current branch

This implementation adopts the core commit 12bc3f3a5c00265f5c04dc28a4e6c76affe326bf from #18280 ([Pipe] Add reliable DataNode completion metric). The complete implementation includes:

waiting during a full FLUSH for working and closing processors and overlapping ordinary asynchronous flushes;
completion tokens, data generations, assigner epochs, and publication-failure epochs;
ordered barrier preservation through the source, processor, and sink queue;
an ordered-commit hook after the sink ACK;
task/source/assigner/committer lifecycle validation; and
a fail-closed completion operator and concurrency tests.

The current branch additionally reuses the expected-DataRegion calculation already present in master and requires:

actual DataRegion source ids == expected DataRegion ids

This prevents a false positive when a DataRegion task fails to initialize or is missing, which could otherwise make an empty or incomplete local task set look complete.

8. Validation results

Spotless: passed;
clean test-compile for the relevant modules: passed;
18 focused barrier, lifecycle, queue, and full-FLUSH concurrency tests: Failures 0, Errors 0, Skipped 0;
English-locale all-reactor test-compile: 52/52 passed;
Chinese-locale all-reactor test-compile: 52/52 passed; and
git diff --check: passed.

The focused tests cover completion generation, fail-closed behavior, membership changes, source replacement, committer replacement, barriers not being merged with heartbeats, event-collection failure, inserts invalidating a flush token, and overlap between a full FLUSH and an ordinary asynchronous flush.

jt2594838 · 2026-09-29T02:11:23Z

  private void collectEvent(final Event event) {
    if (event instanceof EnrichedEvent) {
      final EnrichedEvent enrichedEvent = (EnrichedEvent) event;
      if (!enrichedEvent.increaseReferenceCount(PipeEventCollector.class.getName())) {
+        markDataRegionCompletionInvalid(enrichedEvent);


Why would a skipped event invalidate the completion mark.

jt2594838 · 2026-09-29T02:12:34Z

+      PipeDataNodeSinglePipeMetrics.getInstance()
+          .markDataRegionInvalid(


Is the method name markDataRegionInvalid proper?

jt2594838 · 2026-09-29T02:15:42Z

+  private final long completionSourceId;
+


Is it possible to make the ID not only for completion?

jt2594838 · 2026-09-29T02:25:09Z

+          // A completion barrier is an internal ordering event. It must not be swallowed or
+          // transformed by a user processor.
+          if (((PipeHeartbeatEvent) event).isCompletionBarrier()) {
+            outputEventCollector.collect(event);
+          } else {
+            pipeProcessor.process(event, outputEventCollector);
+          }


May use a higher-level abstraction like Event.isUserVisible or Event.shouldSkipProcessing

jt2594838 · 2026-09-29T02:26:58Z

+  private static final Set<String> COMPLETION_SUPPORTED_SINKS =
+      Set.of(
+          BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR.getPipePluginName(),
+          BuiltinPipePlugin.IOTDB_THRIFT_SSL_CONNECTOR.getPipePluginName(),
+          BuiltinPipePlugin.IOTDB_THRIFT_SYNC_CONNECTOR.getPipePluginName(),
+          BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_CONNECTOR.getPipePluginName(),
+          BuiltinPipePlugin.IOTDB_THRIFT_SINK.getPipePluginName(),
+          BuiltinPipePlugin.IOTDB_THRIFT_SSL_SINK.getPipePluginName(),
+          BuiltinPipePlugin.IOTDB_THRIFT_SYNC_SINK.getPipePluginName(),
+          BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_SINK.getPipePluginName());


Is it possible to support air-gap?

jt2594838 · 2026-09-29T08:19:42Z

+  public void publishDataEventToAssign(final Supplier<PipeRealtimeEvent> eventSupplier) {
+    publishDataEventToAssign(eventSupplier, false);
+  }
+
+  public void publishInsertDataEventToAssign(final Supplier<PipeRealtimeEvent> eventSupplier) {
+    publishDataEventToAssign(eventSupplier, true);
+  }


Explain these.

jt2594838 · 2026-09-29T08:28:15Z

+  /**
+   * Advances the published data generation for a data event that was ignored before publication.
+   * This deliberately keeps the current completion token valid: full-flush close callbacks may be
+   * ignored when no source listens to TsFile events, but their generations still need to be covered
+   * by the barrier for that flush.
+   */
+  public void invalidateCompletion() {
+    invalidateCompletion(false);
+  }
+
+  /**
+   * Advances the published data generation and invalidates the current completion token for an
+   * ignored insert. The insert may create a working TsFile processor that the in-progress full
+   * flush does not cover, so its barrier must not be published.
+   */
+  public void invalidateCompletionBarrier() {
+    invalidateCompletion(true);
+  }


What does a data event refer to?

jt2594838 · 2026-09-29T08:30:07Z

@@ -212,6 +387,9 @@ private void assignToSource(
              }



Is it possible to abstract these as Event.bindSource?

jt2594838 · 2026-09-29T08:52:17Z

+      final Set<TsFileProcessor> targetsWaitingForOrdinaryFlush = new HashSet<>();
+
+      writeLock("syncCloseAllWorkingAndClosingTsFileProcessors");
+      try {
+        final Iterator<TsFileProcessor> iterator = targetsToClose.iterator();
+        while (iterator.hasNext()) {
+          final TsFileProcessor target = iterator.next();
+          final Future<?> future =
+              asyncCloseOneTsFileProcessorForFullFlush(target.isSequence(), target);
+          currentFutures.add(future);
+          if (target.alreadyMarkedClosing()) {
+            closingFutures.add(future);
+            iterator.remove();
+            closedProcessorCount++;
+          } else {
+            targetsWaitingForOrdinaryFlush.add(target);
+          }
+        }
+      } finally {
+        writeUnlock();


What does an ordinary flush refer to?

jt2594838 · 2026-09-29T09:00:19Z

+        && committerKey.equals(
+            generateCommitterKey(
+                committerKey.getPipeName(),
+                committerKey.getCreationTime(),
+                committerKey.getRegionId()));


Explain this

[Pipe] Add reliable DataNode completion metric

dd7bd38

Caideyipi mentioned this pull request Sep 2, 2026

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

Closed

7 tasks done

jt2594838 reviewed Sep 29, 2026

View reviewed changes

Navigation Menu

Repository navigation

[Pipe] Add reliable DataNode completion metric - #18563

[Pipe] Add reliable DataNode completion metric#18563
Caideyipi wants to merge 1 commit into
apache:masterfrom
Caideyipi:cp/pipe-completion-ready-metric

Caideyipi commented Sep 1, 2026 •

edited

Loading

Uh oh!

jt2594838 Sep 29, 2026

Uh oh!

jt2594838 Sep 29, 2026

Uh oh!

jt2594838 Sep 29, 2026

Uh oh!

jt2594838 Sep 29, 2026

Uh oh!

jt2594838 Sep 29, 2026

Uh oh!

jt2594838 Sep 29, 2026

Uh oh!

jt2594838 Sep 29, 2026

Uh oh!

jt2594838 Sep 29, 2026

Uh oh!

jt2594838 Sep 29, 2026

Uh oh!

jt2594838 Sep 29, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

		PipeDataNodeSinglePipeMetrics.getInstance()
		.markDataRegionInvalid(

Conversation

Caideyipi commented Sep 1, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Pipe Completion Metric Design

1. Conclusion

2. Why remaining_event_count cannot simply include "waiting for flush" and "captured TsFile"

2.1 State transitions create counting gaps

2.2 An empty queue does not mean that the sink has confirmed receipt

2.3 FLUSH and concurrent writes have a boundary race

2.4 Lifecycle changes can make old state report completion incorrectly

3. Completion barrier flow

4. Fail-closed conditions

5. Supported scope

6. Correct use of the protocol

7. Relationship to PR #18280 and adaptation on the current branch

8. Validation results

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

Caideyipi commented Sep 1, 2026 •

edited

Loading

2. Why `remaining_event_count` cannot simply include "waiting for flush" and "captured TsFile"