You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ji...@apache.org on 2026/10/15 02:34:32 UTC
(iotdb) branch master updated: [Pipe] Add reliable DataNode completion metric (#18563)
This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 5842fbc6a7b [Pipe] Add reliable DataNode completion metric (#18563)
5842fbc6a7b is described below
commit 5842fbc6a7b0986ad1a1b2b6e056588d4c8685c2
Author: Caideyipi <87...@users.noreply.github.com>
AuthorDate: Thu Oct 15 10:34:24 2026 +0800
[Pipe] Add reliable DataNode completion metric (#18563)
* [Pipe] Add reliable DataNode completion metric
* [Pipe] Address completion metric review feedback
---
.../org/apache/iotdb/pipe/api/event/Event.java | 18 +-
.../iotdb/db/pipe/agent/task/PipeDataNodeTask.java | 6 +
.../db/pipe/agent/task/PipeDataNodeTaskAgent.java | 67 ++++
.../task/builder/PipeDataNodeTaskBuilder.java | 1 +
.../agent/task/connection/PipeEventCollector.java | 23 +-
.../agent/task/stage/PipeTaskProcessorStage.java | 2 +
.../pipe/agent/task/stage/PipeTaskSourceStage.java | 18 +
.../subtask/processor/PipeProcessorSubtask.java | 10 +-
.../sink/PipeRealtimePriorityBlockingQueue.java | 4 +-
.../db/pipe/event/UserDefinedEnrichedEvent.java | 11 +
.../event/common/heartbeat/PipeHeartbeatEvent.java | 102 ++++-
.../event/common/terminate/PipeTerminateEvent.java | 5 +
.../event/realtime/PipeRealtimeEventFactory.java | 4 +
.../overview/PipeDataNodeCompletionOperator.java | 420 +++++++++++++++++++++
.../PipeDataNodeRemainingEventAndTimeOperator.java | 4 +
.../overview/PipeDataNodeSinglePipeMetrics.java | 126 ++++++-
.../source/dataregion/IoTDBDataRegionSource.java | 15 +
.../PipeRealtimeDataRegionHybridSource.java | 20 +-
.../realtime/PipeRealtimeDataRegionLogSource.java | 1 +
.../realtime/PipeRealtimeDataRegionSource.java | 41 +-
.../PipeRealtimeDataRegionTsFileSource.java | 1 +
.../realtime/assigner/PipeDataRegionAssigner.java | 243 +++++++++++-
.../listener/PipeInsertionDataNodeListener.java | 77 ++--
.../iotdb/db/storageengine/StorageEngine.java | 17 +-
.../db/storageengine/dataregion/DataRegion.java | 143 +++++++
.../dataregion/memtable/TsFileProcessor.java | 44 ++-
.../task/connection/PipeEventCollectorTest.java | 4 +-
.../processor/PipeProcessorSubtaskTest.java | 87 +++++
.../common/heartbeat/PipeHeartbeatEventTest.java | 92 +++++
.../PipeDataNodeCompletionOperatorTest.java | 248 ++++++++++++
.../PipeDataNodeSinglePipeMetricsTest.java | 221 +++++++++++
.../airgap/IoTDBDataRegionAirGapSinkTest.java | 34 +-
.../realtime/PipeRealtimeDataRegionSourceTest.java | 25 ++
.../assigner/PipeDataRegionAssignerTest.java | 88 +++++
.../PipeInsertionDataNodeListenerTest.java | 66 ++++
.../storageengine/dataregion/DataRegionTest.java | 141 +++++++
.../agent/plugin/builtin/BuiltinPipePlugin.java | 46 ++-
.../pipe/agent/task/meta/PipeStaticMeta.java | 33 ++
.../commons/pipe/agent/task/meta/PipeTaskMeta.java | 17 +-
.../task/progress/PipeEventCommitManager.java | 15 +
.../iotdb/commons/pipe/event/EnrichedEvent.java | 14 +
.../commons/pipe/event/ProgressReportEvent.java | 5 +
.../iotdb/commons/service/metric/enums/Metric.java | 1 +
.../iotdb/commons/pipe/task/PipeMetaDeSerTest.java | 74 ++++
44 files changed, 2538 insertions(+), 96 deletions(-)
diff --git a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/event/Event.java b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/event/Event.java
index 74ddf9e47d2..b9198a2cdc9 100644
--- a/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/event/Event.java
+++ b/iotdb-api/pipe-api/src/main/java/org/apache/iotdb/pipe/api/event/Event.java
@@ -20,4 +20,20 @@
package org.apache.iotdb.pipe.api.event;
/** This interface is used to abstract events in collaboration tasks. */
-public interface Event {}
+public interface Event {
+
+ /** Whether this event carries internal control flow rather than user data. */
+ default boolean isControlEvent() {
+ return false;
+ }
+
+ /** Whether the processor must forward this event unchanged to preserve internal ordering. */
+ default boolean shouldSkipProcessing() {
+ return false;
+ }
+
+ /** Binds the event to its source instance. Events without source metadata may ignore this. */
+ default void bindSource(final long sourceId) {
+ // Do nothing by default.
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTask.java
index 08158d19e23..6c2fe17acdc 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTask.java
@@ -22,6 +22,8 @@ package org.apache.iotdb.db.pipe.agent.task;
import org.apache.iotdb.commons.pipe.agent.task.PipeTask;
import org.apache.iotdb.commons.pipe.agent.task.stage.PipeTaskStage;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
+import org.apache.iotdb.db.pipe.agent.task.stage.PipeTaskSourceStage;
+import org.apache.iotdb.pipe.api.PipeExtractor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -119,6 +121,10 @@ public class PipeDataNodeTask implements PipeTask {
return pipeName;
}
+ public PipeExtractor getPipeExtractor() {
+ return ((PipeTaskSourceStage) sourceStage).getPipeExtractor();
+ }
+
public boolean isCompleted() {
return isCompleted;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java
index 8bb8e5ddf59..359bdd79c50 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/PipeDataNodeTaskAgent.java
@@ -64,7 +64,10 @@ import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryManager;
import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFileResourceManager;
import org.apache.iotdb.db.pipe.source.dataregion.DataRegionListeningFilter;
+import org.apache.iotdb.db.pipe.source.dataregion.IoTDBDataRegionSource;
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.PipeRealtimeDataRegionSource;
import org.apache.iotdb.db.pipe.source.dataregion.realtime.listener.PipeInsertionDataNodeListener;
+import org.apache.iotdb.db.pipe.source.schemaregion.IoTDBSchemaRegionSource;
import org.apache.iotdb.db.pipe.source.schemaregion.SchemaRegionListeningFilter;
import org.apache.iotdb.db.protocol.client.ConfigNodeClient;
import org.apache.iotdb.db.protocol.client.ConfigNodeClientManager;
@@ -76,6 +79,7 @@ import org.apache.iotdb.db.subscription.agent.SubscriptionAgent;
import org.apache.iotdb.mpp.rpc.thrift.TDataNodeHeartbeatResp;
import org.apache.iotdb.mpp.rpc.thrift.TPipeHeartbeatReq;
import org.apache.iotdb.mpp.rpc.thrift.TPushPipeMetaRespExceptionMessage;
+import org.apache.iotdb.pipe.api.PipeExtractor;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
import org.apache.iotdb.pipe.api.exception.PipeException;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -697,6 +701,69 @@ public class PipeDataNodeTaskAgent extends PipeTaskAgent {
: pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().keySet();
}
+ /**
+ * Returns current realtime DataRegion sources keyed by region ID. The left value is true only for
+ * a supported, running, healthy Pipe whose expected DataRegion tasks are all present and ready. A
+ * false value means completion is unproven; the map may be partial or empty and must not be
+ * interpreted as completed. Runtime state is sampled under the task read lock.
+ */
+ public Pair<Boolean, Map<Integer, PipeRealtimeDataRegionSource>> getPipeCompletionSnapshot(
+ final String pipeName, final long creationTime) {
+ if (!tryReadLockWithTimeOutInMs(10)) {
+ return new Pair<>(false, Collections.emptyMap());
+ }
+
+ try {
+ final PipeMeta pipeMeta = pipeMetaKeeper.getPipeMeta(pipeName, creationTime);
+ if (pipeMeta == null) {
+ return new Pair<>(false, Collections.emptyMap());
+ }
+
+ final PipeStaticMeta staticMeta = pipeMeta.getStaticMeta();
+ boolean supported =
+ staticMeta.supportsCompletionMark()
+ && pipeMeta.getRuntimeMeta().getStatus().get() == PipeStatus.RUNNING
+ && pipeMeta.getRuntimeMeta().getNodeId2PipeRuntimeExceptionMap().isEmpty()
+ && pipeMeta.getRuntimeMeta().getConsensusGroupId2TaskMetaMap().values().stream()
+ .noneMatch(PipeTaskMeta::hasExceptionMessages);
+
+ final Map<Integer, PipeRealtimeDataRegionSource> dataRegionId2Source = new HashMap<>();
+ final Map<Integer, PipeTask> pipeTaskMap = pipeTaskManager.getPipeTasks(staticMeta);
+ if (pipeTaskMap != null) {
+ for (final Map.Entry<Integer, PipeTask> entry : pipeTaskMap.entrySet()) {
+ if (!(entry.getValue() instanceof PipeDataNodeTask)) {
+ supported = false;
+ continue;
+ }
+ final PipeExtractor extractor = ((PipeDataNodeTask) entry.getValue()).getPipeExtractor();
+ if (!(extractor instanceof IoTDBDataRegionSource)) {
+ if (!(extractor instanceof IoTDBSchemaRegionSource)) {
+ supported = false;
+ }
+ continue;
+ }
+ final IoTDBDataRegionSource dataRegionSource = (IoTDBDataRegionSource) extractor;
+ final PipeRealtimeDataRegionSource realtimeSource =
+ dataRegionSource.getRealtimeSourceForCompletion();
+ if (realtimeSource == null) {
+ supported = false;
+ } else {
+ dataRegionId2Source.put(entry.getKey(), realtimeSource);
+ }
+ if (!dataRegionSource.isReadyForCompletion()) {
+ supported = false;
+ }
+ }
+ }
+ if (!dataRegionId2Source.keySet().equals(getExpectedDataRegionIds(pipeMeta))) {
+ supported = false;
+ }
+ return new Pair<>(supported, dataRegionId2Source);
+ } finally {
+ releaseReadLock();
+ }
+ }
+
public boolean hasPipeReleaseRegionRelatedResource(final int consensusGroupId) {
if (!tryReadLockWithTimeOut(10)) {
LOGGER.warn(DataNodePipeMessages.FAILED_TO_CHECK_IF_PIPE_HAS_RELEASE, consensusGroupId);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/builder/PipeDataNodeTaskBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/builder/PipeDataNodeTaskBuilder.java
index ae4ef473998..85d92b72c43 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/builder/PipeDataNodeTaskBuilder.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/builder/PipeDataNodeTaskBuilder.java
@@ -134,6 +134,7 @@ public class PipeDataNodeTaskBuilder {
pipeStaticMeta.getCreationTime(),
blendUserAndSystemParameters(pipeStaticMeta.getProcessorParameters(), pipeTaskMeta),
regionId,
+ sourceStage.getSourceId(),
sourceStage.getEventSupplier(),
sinkStage.getPipeSinkPendingQueue(),
PROCESSOR_EXECUTOR,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
index 53efadc77ed..07ed903b671 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollector.java
@@ -36,6 +36,7 @@ import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertio
import org.apache.iotdb.db.pipe.event.common.tablet.PipeRawTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.terminate.PipeTerminateEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
+import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics;
import org.apache.iotdb.db.pipe.source.schemaregion.IoTDBSchemaRegionSource;
import org.apache.iotdb.db.pipe.source.schemaregion.PipePlanTablePrivilegeParseVisitor;
import org.apache.iotdb.db.pipe.source.schemaregion.PipePlanTreePrivilegeParseVisitor;
@@ -60,6 +61,8 @@ public class PipeEventCollector implements EventCollector {
private final int regionId;
+ private final long sourceId;
+
private final boolean forceTabletFormat;
private final boolean skipParsing;
@@ -76,12 +79,14 @@ public class PipeEventCollector implements EventCollector {
final UnboundedBlockingPendingQueue<Event> pendingQueue,
final long creationTime,
final int regionId,
+ final long sourceId,
final boolean forceTabletFormat,
final boolean skipParsing,
final boolean isUsedInConsensusPipe) {
this.pendingQueue = pendingQueue;
this.creationTime = creationTime;
this.regionId = regionId;
+ this.sourceId = sourceId;
this.forceTabletFormat = forceTabletFormat;
this.skipParsing = skipParsing;
this.isUsedForConsensusPipe = isUsedInConsensusPipe;
@@ -245,7 +250,11 @@ public class PipeEventCollector implements EventCollector {
private void collectEvent(final Event event) {
if (event instanceof EnrichedEvent) {
final EnrichedEvent enrichedEvent = (EnrichedEvent) event;
+ enrichedEvent.bindSource(sourceId);
if (!enrichedEvent.increaseReferenceCount(PipeEventCollector.class.getName())) {
+ // This is a delivery failure: the event's data is no longer safe to access. Unlike a
+ // deliberate pattern/privilege filter, it cannot be covered by a later completion barrier.
+ markDataRegionCompletionInvalid(enrichedEvent);
LOGGER.warn(
DataNodePipeMessages.PIPEEVENTCOLLECTOR_THE_EVENT_IS_ALREADY_RELEASED_SKIPPING, event);
isFailedToIncreaseReferenceCount = true;
@@ -255,6 +264,9 @@ public class PipeEventCollector implements EventCollector {
// Assign a commit id for this event in order to report progress in order.
PipeEventCommitManager.getInstance()
.enrichWithCommitterKeyAndCommitId(enrichedEvent, creationTime, regionId);
+ if (enrichedEvent.needToCommit() && enrichedEvent.getCommitterKey() == null) {
+ markDataRegionCompletionInvalid(enrichedEvent);
+ }
// Assign a rebootTime for iotConsensusV2
enrichedEvent.setRebootTimes(PipeDataNodeAgent.runtime().getRebootTimes());
@@ -284,7 +296,16 @@ public class PipeEventCollector implements EventCollector {
if (pendingQueue.offer(event)) {
collectInvocationCount.incrementAndGet();
- return;
+ } else if (event instanceof EnrichedEvent) {
+ markDataRegionCompletionInvalid((EnrichedEvent) event);
+ }
+ }
+
+ private void markDataRegionCompletionInvalid(final EnrichedEvent event) {
+ if (!event.isControlEvent() && event.getPipeName() != null && regionId >= 0) {
+ PipeDataNodeSinglePipeMetrics.getInstance()
+ .markDataRegionCompletionInvalid(
+ event.getPipeName(), creationTime, regionId, event.getPipeTaskMeta(), sourceId);
}
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskProcessorStage.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskProcessorStage.java
index 85a3ff4cd3c..736a3fbdbd0 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskProcessorStage.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskProcessorStage.java
@@ -67,6 +67,7 @@ public class PipeTaskProcessorStage extends PipeTaskStage {
final long creationTime,
final PipeParameters pipeProcessorParameters,
final int regionId,
+ final long sourceId,
final EventSupplier pipeSourceInputEventSupplier,
final UnboundedBlockingPendingQueue<Event> pipeSinkOutputPendingQueue,
final PipeProcessorSubtaskExecutor executor,
@@ -111,6 +112,7 @@ public class PipeTaskProcessorStage extends PipeTaskStage {
pipeSinkOutputPendingQueue,
creationTime,
regionId,
+ sourceId,
forceTabletFormat,
skipParsing,
isUsedForConsensusPipe);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskSourceStage.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskSourceStage.java
index 460fad472d1..b467d3275bc 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskSourceStage.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/stage/PipeTaskSourceStage.java
@@ -28,6 +28,8 @@ import org.apache.iotdb.commons.pipe.config.plugin.configuraion.PipeTaskRuntimeC
import org.apache.iotdb.commons.pipe.config.plugin.env.PipeTaskSourceRuntimeEnvironment;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
+import org.apache.iotdb.db.pipe.source.dataregion.IoTDBDataRegionSource;
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.PipeRealtimeDataRegionSource;
import org.apache.iotdb.db.storageengine.StorageEngine;
import org.apache.iotdb.pipe.api.PipeExtractor;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
@@ -107,4 +109,20 @@ public class PipeTaskSourceStage extends PipeTaskStage {
public EventSupplier getEventSupplier() {
return pipeExtractor::supply;
}
+
+ public PipeExtractor getPipeExtractor() {
+ return pipeExtractor;
+ }
+
+ public long getSourceId() {
+ if (!(pipeExtractor instanceof IoTDBDataRegionSource)) {
+ return Long.MIN_VALUE;
+ }
+ // Full-FLUSH barriers are delivered by the realtime source. Use that same instance ID for
+ // collector failure reports so a report from a replaced source cannot invalidate its successor.
+ // Historical extraction is checked separately and must be exhausted before completion.
+ final PipeRealtimeDataRegionSource realtimeSource =
+ ((IoTDBDataRegionSource) pipeExtractor).getRealtimeSourceForCompletion();
+ return realtimeSource == null ? Long.MIN_VALUE : realtimeSource.getSourceId();
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
index ca144eca48c..1f065d96087 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtask.java
@@ -173,7 +173,9 @@ public class PipeProcessorSubtask extends PipeReportableSubtask {
}
// event can be supplied after the subtask is closed, so we need to check isClosed here
if (!isClosed.get()) {
- if (event instanceof TabletInsertionEvent) {
+ if (event.shouldSkipProcessing()) {
+ outputEventCollector.collect(event);
+ } else if (event instanceof TabletInsertionEvent) {
if (event instanceof PipeInsertNodeTabletInsertionEvent
&& ((PipeInsertNodeTabletInsertionEvent) event).shouldParse4Privilege()) {
final AtomicReference<Exception> ex = new AtomicReference<>();
@@ -229,8 +231,6 @@ public class PipeProcessorSubtask extends PipeReportableSubtask {
pipeNameWithCreationTime, outputEventCollector.getCollectInvocationCount());
} else if (event instanceof PipeHeartbeatEvent) {
pipeProcessor.process(event, outputEventCollector);
- ((PipeHeartbeatEvent) event).onProcessed();
- PipeProcessorMetrics.getInstance().markPipeHeartbeatEvent(taskID);
} else {
pipeProcessor.process(
event instanceof UserDefinedEnrichedEvent
@@ -238,6 +238,10 @@ public class PipeProcessorSubtask extends PipeReportableSubtask {
: event,
outputEventCollector);
}
+ if (event instanceof PipeHeartbeatEvent) {
+ ((PipeHeartbeatEvent) event).onProcessed();
+ PipeProcessorMetrics.getInstance().markPipeHeartbeatEvent(taskID);
+ }
}
final boolean shouldReport =
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeRealtimePriorityBlockingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeRealtimePriorityBlockingQueue.java
index 8641dbc7867..f224f09425d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeRealtimePriorityBlockingQueue.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeRealtimePriorityBlockingQueue.java
@@ -83,7 +83,9 @@ public class PipeRealtimePriorityBlockingQueue extends UnboundedBlockingPendingQ
return true;
}
- if (event instanceof PipeHeartbeatEvent && super.peekLast() instanceof PipeHeartbeatEvent) {
+ if (event instanceof PipeHeartbeatEvent
+ && !((PipeHeartbeatEvent) event).isCompletionBarrier()
+ && super.peekLast() instanceof PipeHeartbeatEvent) {
// We can NOT keep too many PipeHeartbeatEvent in bufferQueue because they may cause OOM.
((EnrichedEvent) event).decreaseReferenceCount(PipeEventCollector.class.getName(), false);
return false;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/UserDefinedEnrichedEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/UserDefinedEnrichedEvent.java
index 04db68aa3e8..957699e1410 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/UserDefinedEnrichedEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/UserDefinedEnrichedEvent.java
@@ -56,12 +56,23 @@ public class UserDefinedEnrichedEvent extends EnrichedEvent {
enrichedEvent.getEndTime());
this.userDefinedEvent = userDefinedEvent;
this.enrichedEvent = enrichedEvent;
+ bindSource(enrichedEvent.getSourceId());
}
public UserDefinedEvent getUserDefinedEvent() {
return userDefinedEvent;
}
+ @Override
+ public boolean isControlEvent() {
+ return userDefinedEvent.isControlEvent();
+ }
+
+ @Override
+ public boolean shouldSkipProcessing() {
+ return userDefinedEvent.shouldSkipProcessing();
+ }
+
@Override
public boolean internallyIncreaseResourceReferenceCount(final String holderMessage) {
return enrichedEvent.internallyIncreaseResourceReferenceCount(holderMessage);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
index 33697f6d5ab..8c251218c4c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEvent.java
@@ -42,6 +42,9 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
private static final Logger LOGGER = LoggerFactory.getLogger(PipeHeartbeatEvent.class);
private final int dataRegionId;
+ private final boolean completionBarrier;
+ private long assignerEpoch = Long.MIN_VALUE;
+ private long dataGeneration = Long.MIN_VALUE;
private long timePublished;
private long timeAssigned;
@@ -63,9 +66,15 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
private final boolean shouldPrintMessage;
public PipeHeartbeatEvent(final int dataRegionId, final boolean shouldPrintMessage) {
+ this(dataRegionId, shouldPrintMessage, false);
+ }
+
+ public PipeHeartbeatEvent(
+ final int dataRegionId, final boolean shouldPrintMessage, final boolean completionBarrier) {
super(null, 0, null, null, null, null, null, null, true, Long.MIN_VALUE, Long.MAX_VALUE);
this.dataRegionId = dataRegionId;
this.shouldPrintMessage = shouldPrintMessage;
+ this.completionBarrier = completionBarrier;
}
public PipeHeartbeatEvent(
@@ -74,7 +83,11 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
final PipeTaskMeta pipeTaskMeta,
final int dataRegionId,
final long timePublished,
- final boolean shouldPrintMessage) {
+ final boolean shouldPrintMessage,
+ final boolean completionBarrier,
+ final long assignerEpoch,
+ final long dataGeneration,
+ final long sourceId) {
super(
pipeName,
creationTime,
@@ -90,6 +103,13 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
this.dataRegionId = dataRegionId;
this.timePublished = timePublished;
this.shouldPrintMessage = shouldPrintMessage;
+ this.completionBarrier = completionBarrier;
+ this.assignerEpoch = assignerEpoch;
+ this.dataGeneration = dataGeneration;
+ this.sourceId = sourceId;
+ if (completionBarrier && sourceId != Long.MIN_VALUE) {
+ addCompletionOnCommittedHook();
+ }
}
@Override
@@ -136,7 +156,16 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
// Should record PipeTaskMeta, for sometimes HeartbeatEvents should report exceptions.
// Here we ignore parameters `pattern`, `startTime`, and `endTime`.
return new PipeHeartbeatEvent(
- pipeName, creationTime, pipeTaskMeta, dataRegionId, timePublished, shouldPrintMessage);
+ pipeName,
+ creationTime,
+ pipeTaskMeta,
+ dataRegionId,
+ timePublished,
+ shouldPrintMessage,
+ completionBarrier,
+ assignerEpoch,
+ dataGeneration,
+ sourceId);
}
@Override
@@ -160,6 +189,63 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
return shouldPrintMessage;
}
+ public boolean isCompletionBarrier() {
+ return completionBarrier;
+ }
+
+ @Override
+ public boolean isControlEvent() {
+ return true;
+ }
+
+ @Override
+ public boolean shouldSkipProcessing() {
+ return completionBarrier;
+ }
+
+ public void bindCompletionBarrier(final long assignerEpoch, final long dataGeneration) {
+ if (completionBarrier) {
+ this.assignerEpoch = assignerEpoch;
+ this.dataGeneration = dataGeneration;
+ }
+ }
+
+ @Override
+ public void bindSource(final long sourceId) {
+ final boolean shouldAddCompletionHook =
+ completionBarrier
+ && pipeName != null
+ && this.sourceId == Long.MIN_VALUE
+ && sourceId != Long.MIN_VALUE;
+ super.bindSource(sourceId);
+ if (shouldAddCompletionHook) {
+ addCompletionOnCommittedHook();
+ }
+ }
+
+ private void addCompletionOnCommittedHook() {
+ addOnCommittedHook(
+ () ->
+ PipeDataNodeSinglePipeMetrics.getInstance()
+ .markDataRegionCompleted(
+ pipeName,
+ creationTime,
+ dataRegionId,
+ pipeTaskMeta,
+ assignerEpoch,
+ dataGeneration,
+ sourceId,
+ getCommitterKey()));
+ }
+
+ public long getAssignerEpoch() {
+ return assignerEpoch;
+ }
+
+ public long getDataGeneration() {
+ return dataGeneration;
+ }
+
/////////////////////////////// Delay Reporting ///////////////////////////////
public void onPublished() {
@@ -224,10 +310,10 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
/////////////////////////////// For Commit Ordering ///////////////////////////////
- /** {@link PipeHeartbeatEvent}s do not need to be committed in order. */
+ /** Only completion barriers need ordered commit. Periodic heartbeats remain best-effort. */
@Override
public boolean needToCommit() {
- return false;
+ return completionBarrier && Objects.nonNull(pipeName) && sourceId != Long.MIN_VALUE;
}
/////////////////////////////// Object ///////////////////////////////
@@ -278,6 +364,14 @@ public class PipeHeartbeatEvent extends EnrichedEvent {
+ pipeName
+ "', dataRegionId="
+ dataRegionId
+ + ", completionBarrier="
+ + completionBarrier
+ + ", assignerEpoch="
+ + assignerEpoch
+ + ", dataGeneration="
+ + dataGeneration
+ + ", sourceId="
+ + sourceId
+ ", startTime="
+ startTimeMessage
+ ", publishedToAssigned="
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/terminate/PipeTerminateEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/terminate/PipeTerminateEvent.java
index a37a5e525d6..cd454f54626 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/terminate/PipeTerminateEvent.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/common/terminate/PipeTerminateEvent.java
@@ -53,6 +53,11 @@ import java.util.concurrent.atomic.AtomicLong;
*/
public class PipeTerminateEvent extends EnrichedEvent {
+ @Override
+ public boolean isControlEvent() {
+ return true;
+ }
+
private static final Logger LOGGER = LoggerFactory.getLogger(PipeTerminateEvent.class);
private final int dataRegionId;
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEventFactory.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEventFactory.java
index 540589326ff..6784d5a6541 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEventFactory.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEventFactory.java
@@ -62,6 +62,10 @@ public class PipeRealtimeEventFactory {
new PipeHeartbeatEvent(dataRegionId, shouldPrintMessage), null, null);
}
+ public static PipeRealtimeEvent createCompletionBarrierEvent(final int dataRegionId) {
+ return new PipeRealtimeEvent(new PipeHeartbeatEvent(dataRegionId, false, true), null, null);
+ }
+
public static PipeRealtimeEvent createRealtimeEvent(final AbstractDeleteDataNode node) {
PipeDeleteDataNodeEvent deleteDataNodeEvent =
new PipeDeleteDataNodeEvent(node, node.isGeneratedByPipe());
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeCompletionOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeCompletionOperator.java
new file mode 100644
index 00000000000..86c6fa7765a
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeCompletionOperator.java
@@ -0,0 +1,420 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.metric.overview;
+
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
+import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey;
+import org.apache.iotdb.commons.pipe.agent.task.progress.PipeEventCommitManager;
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.PipeRealtimeDataRegionSource;
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeDataRegionAssigner;
+
+import org.apache.tsfile.utils.Pair;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.BooleanSupplier;
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+
+/**
+ * Calculates whether all local DataRegion tasks of a pipe have committed their latest full-flush
+ * completion barriers.
+ *
+ * <p>The operator deliberately fails closed. It takes two task-topology snapshots and two
+ * membership/state snapshots. Any change in task identity, source identity, health, remaining
+ * events, degradation, publication generation, publication failures, or committer incarnation makes
+ * the result incomplete.
+ */
+class PipeDataNodeCompletionOperator {
+
+ private final BooleanSupplier isRemainingEventCountZero;
+ private final Supplier<Pair<Boolean, Map<Integer, PipeRealtimeDataRegionSource>>>
+ taskSnapshotSupplier;
+ private final Predicate<CommitterKey> isCurrentCommitterKey;
+
+ private final ReentrantReadWriteLock membershipLock = new ReentrantReadWriteLock();
+ private final Map<Integer, PipeRealtimeDataRegionSource> registeredSourceMap = new HashMap<>();
+ private final Map<Integer, DataRegionCompletionState> dataRegionId2State = new HashMap<>();
+ private long membershipVersion = 0;
+
+ PipeDataNodeCompletionOperator(
+ final BooleanSupplier isRemainingEventCountZero,
+ final Supplier<Pair<Boolean, Map<Integer, PipeRealtimeDataRegionSource>>>
+ taskSnapshotSupplier) {
+ this(
+ isRemainingEventCountZero,
+ taskSnapshotSupplier,
+ PipeEventCommitManager.getInstance()::isCurrentCommitterKey);
+ }
+
+ PipeDataNodeCompletionOperator(
+ final BooleanSupplier isRemainingEventCountZero,
+ final Supplier<Pair<Boolean, Map<Integer, PipeRealtimeDataRegionSource>>>
+ taskSnapshotSupplier,
+ final Predicate<CommitterKey> isCurrentCommitterKey) {
+ this.isRemainingEventCountZero = isRemainingEventCountZero;
+ this.taskSnapshotSupplier = taskSnapshotSupplier;
+ this.isCurrentCommitterKey = isCurrentCommitterKey;
+ }
+
+ void registerDataRegionSource(final PipeRealtimeDataRegionSource source) {
+ membershipLock.writeLock().lock();
+ try {
+ registeredSourceMap.put(source.getDataRegionId(), source);
+ dataRegionId2State.computeIfPresent(
+ source.getDataRegionId(), (regionId, state) -> state.source == source ? state : null);
+ membershipVersion++;
+ } finally {
+ membershipLock.writeLock().unlock();
+ }
+ }
+
+ void deregisterDataRegionSource(final PipeRealtimeDataRegionSource source) {
+ membershipLock.writeLock().lock();
+ try {
+ if (registeredSourceMap.get(source.getDataRegionId()) == source) {
+ registeredSourceMap.remove(source.getDataRegionId());
+ dataRegionId2State.computeIfPresent(
+ source.getDataRegionId(), (regionId, state) -> state.source == source ? null : state);
+ membershipVersion++;
+ }
+ } finally {
+ membershipLock.writeLock().unlock();
+ }
+ }
+
+ void register(final PipeRealtimeDataRegionSource source, final PipeDataRegionAssigner assigner) {
+ membershipLock.writeLock().lock();
+ try {
+ if (registeredSourceMap.get(source.getDataRegionId()) == source) {
+ dataRegionId2State.put(
+ source.getDataRegionId(),
+ new DataRegionCompletionState(source, assigner, isCurrentCommitterKey));
+ membershipVersion++;
+ }
+ } finally {
+ membershipLock.writeLock().unlock();
+ }
+ }
+
+ void deregister(
+ final PipeRealtimeDataRegionSource source, final PipeDataRegionAssigner assigner) {
+ membershipLock.writeLock().lock();
+ try {
+ final DataRegionCompletionState state = dataRegionId2State.get(source.getDataRegionId());
+ if (state != null && state.matches(source, assigner)) {
+ dataRegionId2State.remove(source.getDataRegionId());
+ membershipVersion++;
+ }
+ } finally {
+ membershipLock.writeLock().unlock();
+ }
+ }
+
+ void markCompleted(
+ final int dataRegionId,
+ final PipeTaskMeta pipeTaskMeta,
+ final long assignerEpoch,
+ final long generation,
+ final long sourceId,
+ final CommitterKey committerKey) {
+ membershipLock.readLock().lock();
+ try {
+ final DataRegionCompletionState state = dataRegionId2State.get(dataRegionId);
+ if (state != null) {
+ state.markCompleted(pipeTaskMeta, assignerEpoch, generation, sourceId, committerKey);
+ }
+ } finally {
+ membershipLock.readLock().unlock();
+ }
+ }
+
+ void markInvalid(final int dataRegionId, final PipeTaskMeta pipeTaskMeta, final long sourceId) {
+ membershipLock.readLock().lock();
+ try {
+ final DataRegionCompletionState state = dataRegionId2State.get(dataRegionId);
+ if (state != null) {
+ state.markInvalid(pipeTaskMeta, sourceId);
+ }
+ } finally {
+ membershipLock.readLock().unlock();
+ }
+ }
+
+ public long getCompletion() {
+ try {
+ if (!isRemainingEventCountZero.getAsBoolean()) {
+ return 0;
+ }
+
+ final Pair<Boolean, Map<Integer, PipeRealtimeDataRegionSource>> firstTaskSnapshot =
+ taskSnapshotSupplier.get();
+ if (!isTaskSnapshotSupported(firstTaskSnapshot)) {
+ return 0;
+ }
+ final Map<Integer, PipeRealtimeDataRegionSource> expectedSources =
+ new HashMap<>(firstTaskSnapshot.getRight());
+
+ final MembershipObservation membershipObservation = observeMembership(expectedSources);
+ if (membershipObservation == null) {
+ return 0;
+ }
+
+ final Pair<Boolean, Map<Integer, PipeRealtimeDataRegionSource>> secondTaskSnapshot =
+ taskSnapshotSupplier.get();
+ if (!isTaskSnapshotSupported(secondTaskSnapshot)
+ || !expectedSources.equals(secondTaskSnapshot.getRight())
+ || !isRemainingEventCountZero.getAsBoolean()) {
+ return 0;
+ }
+
+ return verifyMembership(expectedSources, membershipObservation) ? 1 : 0;
+ } catch (final RuntimeException e) {
+ // A metric must fail closed when task topology or lifecycle is concurrently changing.
+ return 0;
+ }
+ }
+
+ private MembershipObservation observeMembership(
+ final Map<Integer, PipeRealtimeDataRegionSource> expectedSources) {
+ membershipLock.readLock().lock();
+ try {
+ if (!registeredSourceMap.equals(expectedSources)
+ || !dataRegionId2State.keySet().equals(expectedSources.keySet())) {
+ return null;
+ }
+
+ final Map<Integer, StateObservation> stateObservations = new HashMap<>();
+ for (final Map.Entry<Integer, PipeRealtimeDataRegionSource> entry :
+ expectedSources.entrySet()) {
+ final DataRegionCompletionState state = dataRegionId2State.get(entry.getKey());
+ final StateObservation observation =
+ state == null ? null : state.observeCompletion(entry.getValue());
+ if (observation == null) {
+ return null;
+ }
+ stateObservations.put(entry.getKey(), observation);
+ }
+ return new MembershipObservation(membershipVersion, stateObservations);
+ } finally {
+ membershipLock.readLock().unlock();
+ }
+ }
+
+ private boolean verifyMembership(
+ final Map<Integer, PipeRealtimeDataRegionSource> expectedSources,
+ final MembershipObservation membershipObservation) {
+ membershipLock.readLock().lock();
+ try {
+ if (membershipVersion != membershipObservation.membershipVersion
+ || !registeredSourceMap.equals(expectedSources)
+ || !dataRegionId2State.keySet().equals(expectedSources.keySet())) {
+ return false;
+ }
+
+ for (final Map.Entry<Integer, PipeRealtimeDataRegionSource> entry :
+ expectedSources.entrySet()) {
+ final DataRegionCompletionState state = dataRegionId2State.get(entry.getKey());
+ if (state == null
+ || !state.isStillCompleted(
+ entry.getValue(), membershipObservation.stateObservations.get(entry.getKey()))) {
+ return false;
+ }
+ }
+ return true;
+ } finally {
+ membershipLock.readLock().unlock();
+ }
+ }
+
+ private static boolean isTaskSnapshotSupported(
+ final Pair<Boolean, Map<Integer, PipeRealtimeDataRegionSource>> snapshot) {
+ return snapshot != null
+ && Boolean.TRUE.equals(snapshot.getLeft())
+ && snapshot.getRight() != null;
+ }
+
+ private static class DataRegionCompletionState {
+
+ private final PipeRealtimeDataRegionSource source;
+ private final PipeDataRegionAssigner assigner;
+ private final PipeTaskMeta pipeTaskMeta;
+ private final long sourceId;
+ private final long assignerEpoch;
+ private final long initialPublicationFailureEpoch;
+ private final Predicate<CommitterKey> isCurrentCommitterKey;
+ private final AtomicBoolean valid = new AtomicBoolean(true);
+ private final AtomicReference<CommittedBarrier> committedBarrier = new AtomicReference<>();
+
+ private DataRegionCompletionState(
+ final PipeRealtimeDataRegionSource source,
+ final PipeDataRegionAssigner assigner,
+ final Predicate<CommitterKey> isCurrentCommitterKey) {
+ this.source = source;
+ this.assigner = assigner;
+ pipeTaskMeta = source.getPipeTaskMeta();
+ sourceId = source.getSourceId();
+ assignerEpoch = assigner.getAssignerEpoch();
+ initialPublicationFailureEpoch = assigner.getPublicationFailureEpoch();
+ this.isCurrentCommitterKey = isCurrentCommitterKey;
+ }
+
+ private boolean matches(
+ final PipeRealtimeDataRegionSource source, final PipeDataRegionAssigner assigner) {
+ return this.source == source && this.assigner == assigner;
+ }
+
+ private void markCompleted(
+ final PipeTaskMeta pipeTaskMeta,
+ final long assignerEpoch,
+ final long generation,
+ final long sourceId,
+ final CommitterKey committerKey) {
+ if (this.pipeTaskMeta != pipeTaskMeta
+ || this.assignerEpoch != assignerEpoch
+ || this.sourceId != sourceId
+ || committerKey == null) {
+ return;
+ }
+
+ while (true) {
+ final CommittedBarrier previous = committedBarrier.get();
+ if (previous != null && previous.generation > generation) {
+ return;
+ }
+ final CommittedBarrier next = new CommittedBarrier(generation, committerKey);
+ if (committedBarrier.compareAndSet(previous, next)) {
+ return;
+ }
+ }
+ }
+
+ private void markInvalid(final PipeTaskMeta pipeTaskMeta, final long sourceId) {
+ if (this.pipeTaskMeta == pipeTaskMeta && this.sourceId == sourceId) {
+ valid.set(false);
+ }
+ }
+
+ private StateObservation observeCompletion(final PipeRealtimeDataRegionSource expectedSource) {
+ final long sourceStateVersion = source.getCompletionStateVersion();
+ final long exceptionMessageVersion =
+ pipeTaskMeta == null ? Long.MIN_VALUE : pipeTaskMeta.getExceptionMessageVersion();
+ final long publicationFailureEpoch = assigner.getPublicationFailureEpoch();
+ final long publishedGeneration = assigner.getPublishedDataGeneration();
+ final CommittedBarrier barrier = committedBarrier.get();
+
+ if (expectedSource != source
+ || !valid.get()
+ || source.getPipeTaskMeta() != pipeTaskMeta
+ || source.getSourceId() != sourceId
+ || pipeTaskMeta == null
+ || pipeTaskMeta.hasExceptionMessages()
+ || source.isTsFileEpochDegraded()
+ || assigner.getAssignerEpoch() != assignerEpoch
+ || publicationFailureEpoch != initialPublicationFailureEpoch
+ || barrier == null
+ || barrier.generation < publishedGeneration
+ || !isCurrentCommitterKey.test(barrier.committerKey)) {
+ return null;
+ }
+
+ return new StateObservation(
+ this,
+ barrier,
+ sourceStateVersion,
+ exceptionMessageVersion,
+ publicationFailureEpoch,
+ publishedGeneration);
+ }
+
+ private boolean isStillCompleted(
+ final PipeRealtimeDataRegionSource expectedSource, final StateObservation observation) {
+ return observation != null
+ && observation.state == this
+ && expectedSource == source
+ && valid.get()
+ && source.getPipeTaskMeta() == pipeTaskMeta
+ && source.getSourceId() == sourceId
+ && pipeTaskMeta != null
+ && !pipeTaskMeta.hasExceptionMessages()
+ && pipeTaskMeta.getExceptionMessageVersion() == observation.exceptionMessageVersion
+ && !source.isTsFileEpochDegraded()
+ && source.getCompletionStateVersion() == observation.sourceStateVersion
+ && assigner.getAssignerEpoch() == assignerEpoch
+ && assigner.getPublicationFailureEpoch() == initialPublicationFailureEpoch
+ && assigner.getPublicationFailureEpoch() == observation.publicationFailureEpoch
+ && assigner.getPublishedDataGeneration() == observation.publishedGeneration
+ && committedBarrier.get() == observation.barrier
+ && observation.barrier.generation >= observation.publishedGeneration
+ && isCurrentCommitterKey.test(observation.barrier.committerKey);
+ }
+ }
+
+ private static class MembershipObservation {
+
+ private final long membershipVersion;
+ private final Map<Integer, StateObservation> stateObservations;
+
+ private MembershipObservation(
+ final long membershipVersion, final Map<Integer, StateObservation> stateObservations) {
+ this.membershipVersion = membershipVersion;
+ this.stateObservations = stateObservations;
+ }
+ }
+
+ private static class StateObservation {
+
+ private final DataRegionCompletionState state;
+ private final CommittedBarrier barrier;
+ private final long sourceStateVersion;
+ private final long exceptionMessageVersion;
+ private final long publicationFailureEpoch;
+ private final long publishedGeneration;
+
+ private StateObservation(
+ final DataRegionCompletionState state,
+ final CommittedBarrier barrier,
+ final long sourceStateVersion,
+ final long exceptionMessageVersion,
+ final long publicationFailureEpoch,
+ final long publishedGeneration) {
+ this.state = state;
+ this.barrier = barrier;
+ this.sourceStateVersion = sourceStateVersion;
+ this.exceptionMessageVersion = exceptionMessageVersion;
+ this.publicationFailureEpoch = publicationFailureEpoch;
+ this.publishedGeneration = publishedGeneration;
+ }
+ }
+
+ private static class CommittedBarrier {
+
+ private final long generation;
+ private final CommitterKey committerKey;
+
+ private CommittedBarrier(final long generation, final CommitterKey committerKey) {
+ this.generation = generation;
+ this.committerKey = committerKey;
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeRemainingEventAndTimeOperator.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeRemainingEventAndTimeOperator.java
index a79be559dd8..30cfcdba7b5 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeRemainingEventAndTimeOperator.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeRemainingEventAndTimeOperator.java
@@ -198,6 +198,10 @@ public class PipeDataNodeRemainingEventAndTimeOperator extends PipeRemainingOper
dataRegionSources.add(source);
}
+ void deregister(final IoTDBDataRegionSource source) {
+ dataRegionSources.remove(source);
+ }
+
void register(final IoTDBSchemaRegionSource source) {
schemaRegionSources.add(source);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeSinglePipeMetrics.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeSinglePipeMetrics.java
index 0f537fd5800..10c48f7d9ea 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeSinglePipeMetrics.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeSinglePipeMetrics.java
@@ -19,6 +19,8 @@
package org.apache.iotdb.db.pipe.metric.overview;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
+import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey;
import org.apache.iotdb.commons.pipe.agent.task.progress.PipeEventCommitManager;
import org.apache.iotdb.commons.service.metric.enums.Metric;
import org.apache.iotdb.commons.service.metric.enums.Tag;
@@ -27,6 +29,8 @@ import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
import org.apache.iotdb.db.pipe.resource.PipeDataNodeResourceManager;
import org.apache.iotdb.db.pipe.resource.tsfile.PipeTsFileResourceManager;
import org.apache.iotdb.db.pipe.source.dataregion.IoTDBDataRegionSource;
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.PipeRealtimeDataRegionSource;
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeDataRegionAssigner;
import org.apache.iotdb.db.pipe.source.schemaregion.IoTDBSchemaRegionSource;
import org.apache.iotdb.metrics.AbstractMetricService;
import org.apache.iotdb.metrics.metricsets.IMetricSet;
@@ -52,6 +56,8 @@ public class PipeDataNodeSinglePipeMetrics implements IMetricSet {
public final Map<String, PipeDataNodeRemainingEventAndTimeOperator>
remainingEventAndTimeOperatorMap = new ConcurrentHashMap<>();
+ final Map<String, PipeDataNodeCompletionOperator> completionOperatorMap =
+ new ConcurrentHashMap<>();
//////////////////////////// bindTo & unbindFrom (metric framework) ////////////////////////////
@@ -90,6 +96,19 @@ public class PipeDataNodeSinglePipeMetrics implements IMetricSet {
Tag.CREATION_TIME.toString(),
String.valueOf(operator.getCreationTime()));
+ final PipeDataNodeCompletionOperator completionOperator = completionOperatorMap.get(pipeID);
+ if (completionOperator != null) {
+ metricService.createAutoGauge(
+ Metric.PIPE_DATANODE_COMPLETION_READY.toString(),
+ MetricLevel.IMPORTANT,
+ completionOperator,
+ PipeDataNodeCompletionOperator::getCompletion,
+ Tag.NAME.toString(),
+ operator.getPipeName(),
+ Tag.CREATION_TIME.toString(),
+ String.valueOf(operator.getCreationTime()));
+ }
+
// Resources
metricService.createAutoGauge(
Metric.PIPE_FLOATING_MEMORY_USAGE.toString(),
@@ -163,6 +182,15 @@ public class PipeDataNodeSinglePipeMetrics implements IMetricSet {
operator.getPipeName(),
Tag.CREATION_TIME.toString(),
String.valueOf(operator.getCreationTime()));
+ if (completionOperatorMap.containsKey(pipeID)) {
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.PIPE_DATANODE_COMPLETION_READY.toString(),
+ Tag.NAME.toString(),
+ operator.getPipeName(),
+ Tag.CREATION_TIME.toString(),
+ String.valueOf(operator.getCreationTime()));
+ }
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.PIPE_FLOATING_MEMORY_USAGE.toString(),
@@ -201,18 +229,93 @@ public class PipeDataNodeSinglePipeMetrics implements IMetricSet {
public synchronized void register(final IoTDBDataRegionSource source) {
// The metric is global thus the regionId is omitted
final String pipeID = source.getPipeName() + "_" + source.getCreationTime();
- remainingEventAndTimeOperatorMap
- .computeIfAbsent(
+ final PipeDataNodeRemainingEventAndTimeOperator remainingOperator =
+ remainingEventAndTimeOperatorMap.computeIfAbsent(
pipeID,
k ->
new PipeDataNodeRemainingEventAndTimeOperator(
- source.getPipeName(), source.getCreationTime()))
- .register(source);
+ source.getPipeName(), source.getCreationTime()));
+ remainingOperator.register(source);
+ final PipeDataNodeCompletionOperator completionOperator =
+ completionOperatorMap.computeIfAbsent(
+ pipeID,
+ k ->
+ new PipeDataNodeCompletionOperator(
+ () -> remainingOperator.getRemainingNonHeartbeatEvents() == 0,
+ () ->
+ PipeDataNodeAgent.task()
+ .getPipeCompletionSnapshot(
+ source.getPipeName(), source.getCreationTime())));
+ if (source.getRealtimeSourceForCompletion() != null) {
+ completionOperator.registerDataRegionSource(source.getRealtimeSourceForCompletion());
+ }
if (Objects.nonNull(metricService)) {
createMetrics(pipeID);
}
}
+ public void register(
+ final PipeRealtimeDataRegionSource source, final PipeDataRegionAssigner assigner) {
+ final String pipeID = source.getPipeName() + "_" + source.getCreationTime();
+ final PipeDataNodeCompletionOperator operator = completionOperatorMap.get(pipeID);
+ if (operator != null) {
+ operator.register(source, assigner);
+ }
+ }
+
+ public void deregister(final IoTDBDataRegionSource source) {
+ final String pipeID = source.getPipeName() + "_" + source.getCreationTime();
+ final PipeDataNodeRemainingEventAndTimeOperator remainingOperator =
+ remainingEventAndTimeOperatorMap.get(pipeID);
+ if (remainingOperator != null) {
+ remainingOperator.deregister(source);
+ }
+
+ final PipeDataNodeCompletionOperator operator = completionOperatorMap.get(pipeID);
+ if (operator != null && source.getRealtimeSourceForCompletion() != null) {
+ operator.deregisterDataRegionSource(source.getRealtimeSourceForCompletion());
+ }
+ }
+
+ public void deregister(
+ final PipeRealtimeDataRegionSource source, final PipeDataRegionAssigner assigner) {
+ final PipeDataNodeCompletionOperator operator =
+ completionOperatorMap.get(source.getPipeName() + "_" + source.getCreationTime());
+ if (operator != null) {
+ operator.deregister(source, assigner);
+ }
+ }
+
+ public void markDataRegionCompleted(
+ final String pipeName,
+ final long creationTime,
+ final int dataRegionId,
+ final PipeTaskMeta pipeTaskMeta,
+ final long assignerEpoch,
+ final long generation,
+ final long sourceId,
+ final CommitterKey committerKey) {
+ final PipeDataNodeCompletionOperator operator =
+ completionOperatorMap.get(pipeName + "_" + creationTime);
+ if (operator != null) {
+ operator.markCompleted(
+ dataRegionId, pipeTaskMeta, assignerEpoch, generation, sourceId, committerKey);
+ }
+ }
+
+ public void markDataRegionCompletionInvalid(
+ final String pipeName,
+ final long creationTime,
+ final int dataRegionId,
+ final PipeTaskMeta pipeTaskMeta,
+ final long sourceId) {
+ final PipeDataNodeCompletionOperator operator =
+ completionOperatorMap.get(pipeName + "_" + creationTime);
+ if (operator != null) {
+ operator.markInvalid(dataRegionId, pipeTaskMeta, sourceId);
+ }
+ }
+
public synchronized void register(final IoTDBSchemaRegionSource source) {
// The metric is global thus the regionId is omitted
final String pipeID = source.getPipeName() + "_" + source.getCreationTime();
@@ -321,11 +424,11 @@ public class PipeDataNodeSinglePipeMetrics implements IMetricSet {
}
public void decreaseHeartbeatEventCount(final String pipeName, final long creationTime) {
- remainingEventAndTimeOperatorMap
- .computeIfAbsent(
- pipeName + "_" + creationTime,
- k -> new PipeDataNodeRemainingEventAndTimeOperator(pipeName, creationTime))
- .decreaseHeartbeatEventCount();
+ final PipeDataNodeRemainingEventAndTimeOperator operator =
+ remainingEventAndTimeOperatorMap.get(pipeName + "_" + creationTime);
+ if (operator != null) {
+ operator.decreaseHeartbeatEventCount();
+ }
}
public void thawRate(final String pipeID) {
@@ -351,12 +454,11 @@ public class PipeDataNodeSinglePipeMetrics implements IMetricSet {
public synchronized void deregister(final String pipeID) {
if (!remainingEventAndTimeOperatorMap.containsKey(pipeID)) {
LOGGER.warn(DataNodePipeMessages.FAILED_TO_DEREGISTER_PIPE_REMAINING_EVENT_AND, pipeID);
- return;
- }
- if (Objects.nonNull(metricService)) {
+ } else if (Objects.nonNull(metricService)) {
removeMetrics(pipeID);
}
remainingEventAndTimeOperatorMap.remove(pipeID);
+ completionOperatorMap.remove(pipeID);
}
public void markRegionCommit(final String pipeID, final boolean isDataRegion) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java
index 5d677e10047..481ef55485d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/IoTDBDataRegionSource.java
@@ -63,6 +63,7 @@ import javax.annotation.Nonnull;
import java.time.ZoneId;
import java.util.Arrays;
import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import static org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant.EXTRACTOR_DATABASE_KEY;
@@ -141,6 +142,7 @@ public class IoTDBDataRegionSource extends IoTDBSource {
private boolean hasNoExtractionNeed = true;
private boolean shouldExtractDeletion = false;
+ private final AtomicBoolean completionSourceStarted = new AtomicBoolean(false);
@Override
public void validate(final PipeParameterValidator validator) throws Exception {
@@ -562,6 +564,7 @@ public class IoTDBDataRegionSource extends IoTDBSource {
dataRegionIdObject,
() -> startHistoricalExtractorAndRealtimeExtractor(exceptionHolder))) {
rethrowExceptionIfAny(exceptionHolder);
+ completionSourceStarted.set(true);
LOGGER.info(
DataNodePipeMessages.PIPE_STARTED_HISTORICAL_SOURCE_AND_REALTIME_SOURCE,
@@ -637,6 +640,8 @@ public class IoTDBDataRegionSource extends IoTDBSource {
@Override
public void close() throws Exception {
+ completionSourceStarted.set(false);
+ PipeDataNodeSinglePipeMetrics.getInstance().deregister(this);
if (hasNoExtractionNeed || !hasBeenStarted.get()) {
return;
}
@@ -648,6 +653,16 @@ public class IoTDBDataRegionSource extends IoTDBSource {
}
}
+ public PipeRealtimeDataRegionSource getRealtimeSourceForCompletion() {
+ return realtimeSource;
+ }
+
+ public boolean isReadyForCompletion() {
+ return completionSourceStarted.get()
+ && Objects.nonNull(historicalSource)
+ && historicalSource.hasConsumedAll();
+ }
+
//////////////////////////// APIs provided for metric framework ////////////////////////////
public int getHistoricalTsFileInsertionEventCount() {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionHybridSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionHybridSource.java
index 7d3f6be7805..1a927110deb 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionHybridSource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionHybridSource.java
@@ -363,8 +363,11 @@ public class PipeRealtimeDataRegionHybridSource extends PipeRealtimeDataRegionSo
private void markTsFileEpochDegradedUnderLock(final TsFileEpoch tsFileEpoch) {
activeTsFileEpochs.add(tsFileEpoch);
- if (degradedTsFileEpochs.add(tsFileEpoch) && isRegionLevelDowngradingEnabled) {
- regionLevelDegradedTsFileEpochs.offer(tsFileEpoch);
+ if (degradedTsFileEpochs.add(tsFileEpoch)) {
+ markCompletionStateChanged();
+ if (isRegionLevelDowngradingEnabled) {
+ regionLevelDegradedTsFileEpochs.offer(tsFileEpoch);
+ }
}
if (isRegionLevelDowngradingEnabled) {
isRegionLevelDegraded = true;
@@ -522,7 +525,9 @@ public class PipeRealtimeDataRegionHybridSource extends PipeRealtimeDataRegionSo
private void clearTsFileEpochUnderLock(final TsFileEpoch tsFileEpoch) {
activeTsFileEpochs.remove(tsFileEpoch);
- degradedTsFileEpochs.remove(tsFileEpoch);
+ if (degradedTsFileEpochs.remove(tsFileEpoch)) {
+ markCompletionStateChanged();
+ }
regionLevelDegradedTsFileEpochs.remove(tsFileEpoch);
if (regionLevelTailTsFileEpoch == tsFileEpoch) {
regionLevelTailTsFileEpoch = null;
@@ -561,6 +566,11 @@ public class PipeRealtimeDataRegionHybridSource extends PipeRealtimeDataRegionSo
}
}
+ @Override
+ public boolean isTsFileEpochDegraded() {
+ return !degradedTsFileEpochs.isEmpty();
+ }
+
@Override
public void close() throws Exception {
try {
@@ -571,6 +581,9 @@ public class PipeRealtimeDataRegionHybridSource extends PipeRealtimeDataRegionSo
synchronized (regionLevelDowngradingLock) {
clearRegionLevelBufferedEventsUnderLock();
activeTsFileEpochs.clear();
+ if (!degradedTsFileEpochs.isEmpty()) {
+ markCompletionStateChanged();
+ }
degradedTsFileEpochs.clear();
regionLevelDegradedTsFileEpochs.clear();
isRegionLevelDegraded = false;
@@ -890,6 +903,7 @@ public class PipeRealtimeDataRegionHybridSource extends PipeRealtimeDataRegionSo
DataNodePipeMessages.EVENT_CAN_NOT_BE_SUPPLIED_BECAUSE_DATA_IS_LOST,
event.getEvent());
LOGGER.error(errorMessage);
+ markDataRegionCompletionInvalid();
PipeDataNodeAgent.runtime()
.report(pipeTaskMeta, new PipeRuntimeNonCriticalException(errorMessage));
PipeTsFileEpochProgressIndexKeeper.getInstance()
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionLogSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionLogSource.java
index a90441bae2b..21325d82224 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionLogSource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionLogSource.java
@@ -117,6 +117,7 @@ public class PipeRealtimeDataRegionLogSource extends PipeRealtimeDataRegionSourc
DataNodePipeMessages.EVENT_CAN_NOT_BE_SUPPLIED_BECAUSE_DATA_IS_LOST,
realtimeEvent.getEvent());
LOGGER.error(errorMessage);
+ markDataRegionCompletionInvalid();
PipeDataNodeAgent.runtime()
.report(pipeTaskMeta, new PipeRuntimeNonCriticalException(errorMessage));
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
index a678fd0514a..801051b991c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSource.java
@@ -42,6 +42,7 @@ import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
import org.apache.iotdb.db.pipe.event.common.tablet.PipeInsertNodeTabletInsertionEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEvent;
+import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics;
import org.apache.iotdb.db.pipe.metric.source.PipeDataRegionEventCounter;
import org.apache.iotdb.db.pipe.processor.iotconsensusv2.IoTConsensusV2Processor;
import org.apache.iotdb.db.pipe.source.dataregion.DataRegionListeningFilter;
@@ -99,11 +100,15 @@ import static org.apache.iotdb.commons.pipe.source.IoTDBSource.getSkipIfNoPrivil
public abstract class PipeRealtimeDataRegionSource implements PipeExtractor {
private static final Logger LOGGER = LoggerFactory.getLogger(PipeRealtimeDataRegionSource.class);
+ private static final AtomicLong SOURCE_ID_GENERATOR = new AtomicLong(0);
protected String pipeName;
protected long creationTime;
protected int dataRegionId = -1;
protected PipeTaskMeta pipeTaskMeta;
+ // A lifecycle identity for this source instance, also attached to its events for attribution.
+ private final long sourceId = SOURCE_ID_GENERATOR.incrementAndGet();
+ private final AtomicLong completionStateVersion = new AtomicLong(0);
protected boolean shouldExtractInsertion;
protected boolean shouldExtractDeletion;
@@ -454,13 +459,15 @@ public abstract class PipeRealtimeDataRegionSource implements PipeExtractor {
protected void extractHeartbeat(final PipeRealtimeEvent event) {
// Record the pending queue size before trying to put heartbeatEvent into queue
- ((PipeHeartbeatEvent) event.getEvent()).recordExtractorQueueSize(pendingQueue);
+ final PipeHeartbeatEvent heartbeatEvent = (PipeHeartbeatEvent) event.getEvent();
+ heartbeatEvent.recordExtractorQueueSize(pendingQueue);
final Event lastEvent = pendingQueue.peekLast();
- if (lastEvent instanceof PipeRealtimeEvent
+ if (!heartbeatEvent.isCompletionBarrier()
+ && lastEvent instanceof PipeRealtimeEvent
&& ((PipeRealtimeEvent) lastEvent).getEvent() instanceof PipeHeartbeatEvent
&& (((PipeHeartbeatEvent) ((PipeRealtimeEvent) lastEvent).getEvent()).isShouldPrintMessage()
- || !((PipeHeartbeatEvent) event.getEvent()).isShouldPrintMessage())) {
+ || !heartbeatEvent.isShouldPrintMessage())) {
// If the last event in the pending queue is a heartbeat event, we should not extract any more
// heartbeat events to avoid OOM when the pipe is stopped.
// Besides, the printable event has higher priority to stay in queue to enable metrics report.
@@ -478,7 +485,9 @@ public abstract class PipeRealtimeDataRegionSource implements PipeExtractor {
// yet
while (true) {
final PipeRealtimeEvent lastEvent = ((PipeRealtimeEvent) pendingQueue.peekLast());
- if (lastEvent == null || !(lastEvent.getEvent() instanceof PipeHeartbeatEvent)) {
+ if (lastEvent == null
+ || !(lastEvent.getEvent() instanceof PipeHeartbeatEvent)
+ || ((PipeHeartbeatEvent) lastEvent.getEvent()).isCompletionBarrier()) {
break;
}
final PipeRealtimeEvent droppedEvent = (PipeRealtimeEvent) pendingQueue.pollLast();
@@ -565,6 +574,7 @@ public abstract class PipeRealtimeDataRegionSource implements PipeExtractor {
DataNodePipeMessages.EVENT_CAN_NOT_BE_SUPPLIED_BECAUSE_DATA_IS_LOST,
event.getEvent());
LOGGER.error(errorMessage);
+ markDataRegionCompletionInvalid();
PipeDataNodeAgent.runtime()
.report(pipeTaskMeta, new PipeRuntimeNonCriticalException(errorMessage));
return null;
@@ -619,6 +629,29 @@ public abstract class PipeRealtimeDataRegionSource implements PipeExtractor {
return dataRegionId;
}
+ public final long getSourceId() {
+ return sourceId;
+ }
+
+ public final long getCompletionStateVersion() {
+ return completionStateVersion.get();
+ }
+
+ protected final void markCompletionStateChanged() {
+ completionStateVersion.incrementAndGet();
+ }
+
+ protected final void markDataRegionCompletionInvalid() {
+ PipeDataNodeSinglePipeMetrics.getInstance()
+ .markDataRegionCompletionInvalid(
+ pipeName, creationTime, dataRegionId, pipeTaskMeta, sourceId);
+ }
+
+ /** Whether this source is waiting for a TsFile to recover discarded tablet events. */
+ public boolean isTsFileEpochDegraded() {
+ return false;
+ }
+
public final long getRealtimeDataExtractionStartTime() {
return realtimeDataExtractionStartTime;
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionTsFileSource.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionTsFileSource.java
index 1e7a000c8c8..14db05e678c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionTsFileSource.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionTsFileSource.java
@@ -236,6 +236,7 @@ public class PipeRealtimeDataRegionTsFileSource extends PipeRealtimeDataRegionSo
DataNodePipeMessages.EVENT_CAN_NOT_BE_SUPPLIED_BECAUSE_DATA_IS_LOST,
realtimeEvent.getEvent());
LOGGER.error(errorMessage);
+ markDataRegionCompletionInvalid();
PipeDataNodeAgent.runtime()
.report(pipeTaskMeta, new PipeRuntimeNonCriticalException(errorMessage));
PipeTsFileEpochProgressIndexKeeper.getInstance()
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java
index 4c717da3c53..6e308b5f159 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssigner.java
@@ -32,6 +32,7 @@ import org.apache.iotdb.db.pipe.event.common.heartbeat.PipeHeartbeatEvent;
import org.apache.iotdb.db.pipe.event.common.tsfile.PipeTsFileInsertionEvent;
import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEvent;
import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEventFactory;
+import org.apache.iotdb.db.pipe.metric.overview.PipeDataNodeSinglePipeMetrics;
import org.apache.iotdb.db.pipe.metric.source.PipeAssignerMetrics;
import org.apache.iotdb.db.pipe.metric.source.PipeDataRegionEventCounter;
import org.apache.iotdb.db.pipe.source.dataregion.realtime.PipeRealtimeDataRegionSource;
@@ -49,10 +50,14 @@ import org.slf4j.LoggerFactory;
import java.io.Closeable;
import java.util.Objects;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.function.Supplier;
public class PipeDataRegionAssigner implements Closeable {
private static final Logger LOGGER = LoggerFactory.getLogger(PipeDataRegionAssigner.class);
+ private static final AtomicLong ASSIGNER_EPOCH_GENERATOR = new AtomicLong(0);
/**
* The {@link PipeDataRegionMatcher} is used to match the event with the source based on the
@@ -64,6 +69,16 @@ public class PipeDataRegionAssigner implements Closeable {
private final DisruptorQueue disruptor;
private final int dataRegionId;
+ // Distinguishes this assigner from a replacement, whose generations start over at zero.
+ private final long assignerEpoch = ASSIGNER_EPOCH_GENERATOR.incrementAndGet();
+ // Advances for every data mutation/publication attempt, including events ignored by listeners.
+ private final AtomicLong publishedDataGeneration = new AtomicLong(0);
+ // Records delivery failures that a later barrier cannot prove have been recovered.
+ private final AtomicLong publicationFailureEpoch = new AtomicLong(0);
+ // Invalidates a full-FLUSH token when an insert may create a processor outside its snapshot.
+ private final AtomicLong completionInvalidationEpoch = new AtomicLong(0);
+ // Serializes generation/token changes with publication, so a barrier cannot overtake data.
+ private final ReentrantLock publicationLock = new ReentrantLock();
private Boolean isTableModel;
@@ -94,42 +109,199 @@ public class PipeDataRegionAssigner implements Closeable {
}
public void publishToAssign(final PipeRealtimeEvent event) {
+ publicationLock.lock();
+ try {
+ final EnrichedEvent innerEvent = event.getEvent();
+ final boolean isDataEvent = !innerEvent.isControlEvent();
+ if (isDataEvent) {
+ publishedDataGeneration.incrementAndGet();
+ }
+ publishToAssignInternal(event, isDataEvent);
+ } finally {
+ publicationLock.unlock();
+ }
+ }
+
+ /**
+ * Publishes a TsFile-close or deletion event. Count the attempt before event creation so a failed
+ * supplier still prevents completion. A close callback belongs to the current flush and must not
+ * invalidate its token. Data events carry insert/tablet, TsFile, or deletion data; control events
+ * such as heartbeats and progress reports do not advance the data generation.
+ */
+ public void publishDataEventToAssign(final Supplier<PipeRealtimeEvent> eventSupplier) {
+ publishDataEventToAssign(eventSupplier, false);
+ }
+
+ /**
+ * Publishes an insert and also invalidates the current full-FLUSH token: the insert may create a
+ * working processor outside that flush's captured set. Invalidation and publication share a lock.
+ */
+ public void publishInsertDataEventToAssign(final Supplier<PipeRealtimeEvent> eventSupplier) {
+ publishDataEventToAssign(eventSupplier, true);
+ }
+
+ private void publishDataEventToAssign(
+ final Supplier<PipeRealtimeEvent> eventSupplier, final boolean invalidateCompletionBarrier) {
+ publicationLock.lock();
+ try {
+ publishedDataGeneration.incrementAndGet();
+ if (invalidateCompletionBarrier) {
+ completionInvalidationEpoch.incrementAndGet();
+ }
+ final PipeRealtimeEvent event;
+ try {
+ event = eventSupplier.get();
+ } catch (final RuntimeException | Error e) {
+ markPublicationFailed();
+ throw e;
+ }
+ if (event != null) {
+ publishToAssignInternal(event, true);
+ } else {
+ markPublicationFailed();
+ }
+ } finally {
+ publicationLock.unlock();
+ }
+ }
+
+ private void publishToAssignInternal(final PipeRealtimeEvent event, final boolean isDataEvent) {
+ final EnrichedEvent innerEvent = event.getEvent();
if (!event.increaseReferenceCount(PipeDataRegionAssigner.class.getName())) {
+ if (isDataEvent) {
+ markPublicationFailed();
+ }
LOGGER.warn(DataNodePipeMessages.THE_REFERENCE_COUNT_OF_THE_REALTIME_EVENT, event);
return;
}
- final EnrichedEvent innerEvent = event.getEvent();
eventCounter.increaseEventCount(innerEvent);
if (innerEvent instanceof PipeHeartbeatEvent) {
((PipeHeartbeatEvent) innerEvent).onPublished();
}
- synchronized (this) {
- if (disruptor.isClosed()) {
- onAssignedHook(event);
- return;
- }
- inFlightPublishCount++;
- }
-
+ boolean shouldReleaseDirectly = false;
boolean isPublished = false;
try {
- isPublished = disruptor.publishOrDrop(event);
- } finally {
+ if (innerEvent instanceof PipeHeartbeatEvent) {
+ final PipeHeartbeatEvent heartbeatEvent = (PipeHeartbeatEvent) innerEvent;
+ if (heartbeatEvent.isCompletionBarrier()) {
+ heartbeatEvent.bindCompletionBarrier(assignerEpoch, publishedDataGeneration.get());
+ }
+ }
+
synchronized (this) {
- inFlightPublishCount--;
- if (inFlightPublishCount == 0) {
- notifyAll();
+ if (disruptor.isClosed()) {
+ shouldReleaseDirectly = true;
+ } else {
+ inFlightPublishCount++;
}
}
+
+ if (!shouldReleaseDirectly) {
+ try {
+ isPublished = disruptor.publishOrDrop(event);
+ } finally {
+ synchronized (this) {
+ inFlightPublishCount--;
+ if (inFlightPublishCount == 0) {
+ notifyAll();
+ }
+ }
+ }
+ }
+ } catch (final RuntimeException | Error e) {
+ if (isDataEvent) {
+ markPublicationFailed();
+ }
+ throw e;
}
- if (!isPublished) {
+ if (shouldReleaseDirectly || !isPublished) {
+ if (isDataEvent) {
+ markPublicationFailed();
+ }
onAssignedHook(event);
}
}
+ private void markPublicationFailed() {
+ publicationFailureEpoch.incrementAndGet();
+ }
+
+ /**
+ * 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);
+ }
+
+ private void invalidateCompletion(final boolean invalidateCompletionBarrier) {
+ publicationLock.lock();
+ try {
+ publishedDataGeneration.incrementAndGet();
+ if (invalidateCompletionBarrier) {
+ completionInvalidationEpoch.incrementAndGet();
+ }
+ } finally {
+ publicationLock.unlock();
+ }
+ }
+
+ /**
+ * Atomically advances the generation, invalidates any prior token, and returns the token for a
+ * new full flush. The corresponding completion barrier is accepted only if no insert invalidates
+ * this token before the flush finishes.
+ */
+ public CompletionToken invalidateCompletionAndGetToken() {
+ publicationLock.lock();
+ try {
+ publishedDataGeneration.incrementAndGet();
+ return new CompletionToken(assignerEpoch, completionInvalidationEpoch.incrementAndGet());
+ } finally {
+ publicationLock.unlock();
+ }
+ }
+
+ public boolean publishCompletionBarrier(final CompletionToken token) {
+ publicationLock.lock();
+ try {
+ if (token == null
+ || token.assignerEpoch != assignerEpoch
+ || token.completionInvalidationEpoch != completionInvalidationEpoch.get()) {
+ return false;
+ }
+ publishToAssignInternal(
+ PipeRealtimeEventFactory.createCompletionBarrierEvent(dataRegionId), false);
+ return true;
+ } finally {
+ publicationLock.unlock();
+ }
+ }
+
+ public static final class CompletionToken {
+
+ private final long assignerEpoch;
+ private final long completionInvalidationEpoch;
+
+ private CompletionToken(final long assignerEpoch, final long completionInvalidationEpoch) {
+ this.assignerEpoch = assignerEpoch;
+ this.completionInvalidationEpoch = completionInvalidationEpoch;
+ }
+ }
+
private void onAssignedHook(final PipeRealtimeEvent realtimeEvent) {
realtimeEvent.gcSchemaInfo();
realtimeEvent.decreaseReferenceCount(PipeDataRegionAssigner.class.getName(), false);
@@ -148,6 +320,17 @@ public class PipeDataRegionAssigner implements Closeable {
return;
}
+ try {
+ assignToSourceInternal(event);
+ } catch (final RuntimeException | Error e) {
+ if (!event.getEvent().isControlEvent()) {
+ markPublicationFailed();
+ }
+ throw e;
+ }
+ }
+
+ private void assignToSourceInternal(final PipeRealtimeEvent event) {
final Pair<Set<PipeRealtimeDataRegionSource>, Set<PipeRealtimeDataRegionSource>>
matchedAndUnmatched = matcher.match(event);
@@ -164,7 +347,9 @@ public class PipeDataRegionAssigner implements Closeable {
new ProgressReportEvent(
source.getPipeName(), source.getCreationTime(), source.getPipeTaskMeta());
reportEvent.bindProgressIndex(event.getProgressIndex());
+ reportEvent.bindSource(source.getSourceId());
if (!reportEvent.increaseReferenceCount(PipeDataRegionAssigner.class.getName())) {
+ markPublicationFailed();
LOGGER.warn(
DataNodePipeMessages.THE_REFERENCE_COUNT_OF_THE_EVENT_CANNOT, reportEvent);
return;
@@ -188,6 +373,8 @@ public class PipeDataRegionAssigner implements Closeable {
source.getRealtimeDataExtractionEndTime());
final EnrichedEvent innerEvent = copiedEvent.getEvent();
+ innerEvent.bindSource(source.getSourceId());
+
if (innerEvent instanceof PipeTsFileInsertionEvent) {
final PipeTsFileInsertionEvent tsFileInsertionEvent =
(PipeTsFileInsertionEvent) innerEvent;
@@ -212,6 +399,9 @@ public class PipeDataRegionAssigner implements Closeable {
}
if (!copiedEvent.increaseReferenceCount(PipeDataRegionAssigner.class.getName())) {
+ if (!event.getEvent().isControlEvent()) {
+ markPublicationFailed();
+ }
LOGGER.warn(
DataNodePipeMessages.THE_REFERENCE_COUNT_OF_THE_EVENT_CANNOT, copiedEvent);
return;
@@ -234,7 +424,9 @@ public class PipeDataRegionAssigner implements Closeable {
new ProgressReportEvent(
source.getPipeName(), source.getCreationTime(), source.getPipeTaskMeta());
reportEvent.bindProgressIndex(event.getProgressIndex());
+ reportEvent.bindSource(source.getSourceId());
if (!reportEvent.increaseReferenceCount(PipeDataRegionAssigner.class.getName())) {
+ markPublicationFailed();
LOGGER.warn(
DataNodePipeMessages.THE_REFERENCE_COUNT_OF_THE_EVENT_CANNOT, reportEvent);
return;
@@ -245,7 +437,13 @@ public class PipeDataRegionAssigner implements Closeable {
}
public synchronized void startAssignTo(final PipeRealtimeDataRegionSource source) {
- matcher.register(source);
+ PipeDataNodeSinglePipeMetrics.getInstance().register(source, this);
+ try {
+ matcher.register(source);
+ } catch (final RuntimeException | Error e) {
+ PipeDataNodeSinglePipeMetrics.getInstance().deregister(source, this);
+ throw e;
+ }
if (source.isNeedListenToTsFile()) {
listenToTsFileSourceCount++;
}
@@ -257,6 +455,7 @@ public class PipeDataRegionAssigner implements Closeable {
public synchronized void stopAssignTo(final PipeRealtimeDataRegionSource source) {
matcher.deregister(source);
+ PipeDataNodeSinglePipeMetrics.getInstance().deregister(source, this);
if (source.isNeedListenToTsFile()) {
listenToTsFileSourceCount--;
}
@@ -349,4 +548,16 @@ public class PipeDataRegionAssigner implements Closeable {
public Boolean isTableModel() {
return isTableModel;
}
+
+ public long getAssignerEpoch() {
+ return assignerEpoch;
+ }
+
+ public long getPublishedDataGeneration() {
+ return publishedDataGeneration.get();
+ }
+
+ public long getPublicationFailureEpoch() {
+ return publicationFailureEpoch.get();
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeInsertionDataNodeListener.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeInsertionDataNodeListener.java
index 7e3c25e8062..9c2a4bee090 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeInsertionDataNodeListener.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeInsertionDataNodeListener.java
@@ -26,6 +26,7 @@ import org.apache.iotdb.db.pipe.consensus.deletion.DeletionResourceManager;
import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEventFactory;
import org.apache.iotdb.db.pipe.source.dataregion.realtime.PipeRealtimeDataRegionSource;
import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeDataRegionAssigner;
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeDataRegionAssigner.CompletionToken;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.AbstractDeleteDataNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode;
@@ -34,6 +35,7 @@ import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicReference;
/**
* PipeInsertionEventListener is a singleton in each data node.
@@ -101,14 +103,19 @@ public class PipeInsertionDataNodeListener {
final boolean isLoaded) {
final PipeDataRegionAssigner assigner = dataRegionId2Assigner.get(dataRegionId);
+ if (assigner == null) {
+ return;
+ }
// only events from registered data region with tsfile listeners will be extracted
- if (assigner == null || !assigner.shouldListenToTsFile()) {
+ if (!assigner.shouldListenToTsFile()) {
+ assigner.invalidateCompletion();
return;
}
- assigner.publishToAssign(
- PipeRealtimeEventFactory.createRealtimeEvent(
- assigner.isTableModel(), databaseName, tsFileResource, isLoaded));
+ assigner.publishDataEventToAssign(
+ () ->
+ PipeRealtimeEventFactory.createRealtimeEvent(
+ assigner.isTableModel(), databaseName, tsFileResource, isLoaded));
}
public void listenToInsertNode(
@@ -118,14 +125,19 @@ public class PipeInsertionDataNodeListener {
final TsFileResource tsFileResource) {
final PipeDataRegionAssigner assigner = dataRegionId2Assigner.get(dataRegionId);
+ if (assigner == null) {
+ return;
+ }
// only events from registered data region with insert listeners will be extracted
- if (assigner == null || !assigner.shouldListenToInsertNode()) {
+ if (!assigner.shouldListenToInsertNode()) {
+ assigner.invalidateCompletionBarrier();
return;
}
- assigner.publishToAssign(
- PipeRealtimeEventFactory.createRealtimeEvent(
- assigner.isTableModel(), databaseName, insertNode, tsFileResource));
+ assigner.publishInsertDataEventToAssign(
+ () ->
+ PipeRealtimeEventFactory.createRealtimeEvent(
+ assigner.isTableModel(), databaseName, insertNode, tsFileResource));
}
public DeletionResource listenToDeleteData(
@@ -137,25 +149,25 @@ public class PipeInsertionDataNodeListener {
&& ((RelationalDeleteDataNode) node).getModEntries().isEmpty()) {
return null;
}
+ final AtomicReference<DeletionResource> deletionResourceHolder = new AtomicReference<>();
+ assigner.publishDataEventToAssign(
+ () -> {
+ // Register a deletionResource and return it to DataRegion. A deleteNode generated by a
+ // remote consensus leader should not be persisted to DAL.
+ final DeletionResourceManager manager = DeletionResourceManager.getInstance(regionId);
+ if (Objects.nonNull(manager)
+ && DeletionResource.isDeleteNodeGeneratedInLocalByIoTV2(node)) {
+ final DeletionResource deletionResource = manager.registerDeletionResource(node);
+ deletionResourceHolder.set(deletionResource);
+ // If persistence failed, skip this event to stay consistent with storage engine.
+ if (deletionResource.waitForResult() == DeletionResource.Status.FAILURE) {
+ return null;
+ }
+ }
+ return PipeRealtimeEventFactory.createRealtimeEvent(node);
+ });
- final DeletionResource deletionResource;
- // register a deletionResource and return it to DataRegion
- final DeletionResourceManager manager = DeletionResourceManager.getInstance(regionId);
- // deleteNode generated by remote consensus leader shouldn't be persisted to DAL.
- if (Objects.nonNull(manager) && DeletionResource.isDeleteNodeGeneratedInLocalByIoTV2(node)) {
- deletionResource = manager.registerDeletionResource(node);
- // if persist failed, skip sending/publishing this event to keep consistency with the
- // behavior of storage engine.
- if (deletionResource.waitForResult() == DeletionResource.Status.FAILURE) {
- return deletionResource;
- }
- } else {
- deletionResource = null;
- }
-
- assigner.publishToAssign(PipeRealtimeEventFactory.createRealtimeEvent(node));
-
- return deletionResource;
+ return deletionResourceHolder.get();
}
public void listenToHeartbeat(final boolean shouldPrintMessage) {
@@ -165,6 +177,19 @@ public class PipeInsertionDataNodeListener {
PipeRealtimeEventFactory.createRealtimeEvent(key, shouldPrintMessage)));
}
+ public CompletionToken invalidateDataRegionCompletion(final int dataRegionId) {
+ final PipeDataRegionAssigner assigner = dataRegionId2Assigner.get(dataRegionId);
+ return assigner == null ? null : assigner.invalidateCompletionAndGetToken();
+ }
+
+ public void listenToCompletionBarrier(
+ final int dataRegionId, final CompletionToken completionToken) {
+ final PipeDataRegionAssigner assigner = dataRegionId2Assigner.get(dataRegionId);
+ if (assigner != null) {
+ assigner.publishCompletionBarrier(completionToken);
+ }
+ }
+
public boolean isEmpty() {
return dataRegionId2Assigner.isEmpty();
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
index 93ec1f560a6..72bdf195998 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java
@@ -59,6 +59,8 @@ import org.apache.iotdb.db.exception.load.LoadReadOnlyException;
import org.apache.iotdb.db.exception.runtime.StorageEngineFailureException;
import org.apache.iotdb.db.i18n.StorageEngineMessages;
import org.apache.iotdb.db.pipe.agent.PipeDataNodeAgent;
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeDataRegionAssigner.CompletionToken;
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.listener.PipeInsertionDataNodeListener;
import org.apache.iotdb.db.queryengine.plan.analyze.cache.schema.DataNodeTTLCache;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.load.LoadTsFilePieceNode;
import org.apache.iotdb.db.queryengine.plan.scheduler.load.LoadTsFileScheduler;
@@ -533,7 +535,7 @@ public class StorageEngine implements IService {
tasks.add(
cachedThreadPool.submit(
() -> {
- dataRegion.syncCloseAllWorkingTsFileProcessors();
+ syncCloseAllProcessorAndPublishCompletionBarrier(dataRegion);
return null;
}));
}
@@ -564,7 +566,7 @@ public class StorageEngine implements IService {
tasks.add(
cachedThreadPool.submit(
() -> {
- dataRegion.syncCloseAllWorkingTsFileProcessors();
+ syncCloseAllProcessorAndPublishCompletionBarrier(dataRegion);
return null;
}));
}
@@ -579,7 +581,7 @@ public class StorageEngine implements IService {
tasks.add(
cachedThreadPool.submit(
() -> {
- dataRegion.syncCloseAllWorkingTsFileProcessors();
+ syncCloseAllProcessorAndPublishCompletionBarrier(dataRegion);
return null;
}));
}
@@ -620,6 +622,15 @@ public class StorageEngine implements IService {
}
}
+ private void syncCloseAllProcessorAndPublishCompletionBarrier(final DataRegion dataRegion)
+ throws InterruptedException, ExecutionException {
+ final PipeInsertionDataNodeListener listener = PipeInsertionDataNodeListener.getInstance();
+ final int dataRegionId = dataRegion.getDataRegionId();
+ final CompletionToken completionToken = listener.invalidateDataRegionCompletion(dataRegionId);
+ dataRegion.syncCloseAllWorkingAndClosingTsFileProcessors();
+ listener.listenToCompletionBarrier(dataRegionId, completionToken);
+ }
+
/**
* merge all databases.
*
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
index deaefdd2526..46952ee87fa 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java
@@ -2569,6 +2569,81 @@ public class DataRegion implements IDataRegionForQuery {
}
}
+ /**
+ * Closes all working TsFile processors and waits for processors that were already closing.
+ *
+ * <p>This stronger variant is used by a full flush before a Pipe completion barrier is published.
+ * Unlike {@link #syncCloseAllWorkingTsFileProcessors()}, failures are propagated to the caller so
+ * a failed flush can never publish a completion barrier.
+ */
+ public void syncCloseAllWorkingAndClosingTsFileProcessors()
+ throws InterruptedException, ExecutionException {
+ final Set<TsFileProcessor> targets = new HashSet<>();
+ final Set<TsFileProcessor> targetsToClose = new HashSet<>();
+ final List<Future<?>> closingFutures = new ArrayList<>();
+
+ writeLock("syncCloseAllWorkingAndClosingTsFileProcessors");
+ try {
+ targets.addAll(closingSequenceTsFileProcessor);
+ targets.addAll(closingUnSequenceTsFileProcessor);
+ for (final TsFileProcessor target : targets) {
+ closingFutures.add(target.getCloseFuture());
+ }
+
+ targetsToClose.addAll(workSequenceTsFileProcessors.values());
+ targetsToClose.addAll(workUnsequenceTsFileProcessors.values());
+ targets.addAll(targetsToClose);
+ } finally {
+ writeUnlock();
+ }
+
+ int closedProcessorCount = 0;
+ while (!targetsToClose.isEmpty()) {
+ final List<Future<?>> currentFutures = new ArrayList<>();
+ // An ordinary flush is TsFileProcessor.asyncFlush(): it flushes memtables without closing
+ // the file. Await that flush outside the region lock, then retry the close for the full
+ // FLUSH.
+ 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();
+ }
+
+ for (final Future<?> future : currentFutures) {
+ if (future != null) {
+ future.get();
+ }
+ }
+ for (final TsFileProcessor target : targetsWaitingForOrdinaryFlush) {
+ target.waitUntilFlushManagerReleased();
+ }
+ }
+
+ WritingMetrics.getInstance().recordManualFlushMemTableCount(closedProcessorCount);
+ for (final Future<?> future : closingFutures) {
+ if (future != null) {
+ future.get();
+ }
+ }
+ waitClosingTsFileProcessorFinished(targets);
+ }
+
public void syncCloseWorkingTsFileProcessors(boolean sequence) {
try {
writeLock("syncCloseWorkingTsFileProcessors");
@@ -2633,6 +2708,27 @@ public class DataRegion implements IDataRegionForQuery {
}
}
+ private void waitClosingTsFileProcessorFinished(final Set<TsFileProcessor> targets)
+ throws InterruptedException {
+ while (containsClosingTarget(targets)) {
+ synchronized (closeStorageGroupCondition) {
+ if (containsClosingTarget(targets)) {
+ closeStorageGroupCondition.wait(60_000);
+ }
+ }
+ }
+ }
+
+ private boolean containsClosingTarget(final Set<TsFileProcessor> targets) {
+ for (final TsFileProcessor target : targets) {
+ if (closingSequenceTsFileProcessor.contains(target)
+ || closingUnSequenceTsFileProcessor.contains(target)) {
+ return true;
+ }
+ }
+ return false;
+ }
+
/** close all working tsfile processors */
public List<Future<?>> asyncCloseAllWorkingTsFileProcessors() {
writeLock("asyncCloseAllWorkingTsFileProcessors");
@@ -2661,6 +2757,53 @@ public class DataRegion implements IDataRegionForQuery {
return futures;
}
+ private Future<?> asyncCloseOneTsFileProcessorForFullFlush(
+ final boolean sequence, final TsFileProcessor tsFileProcessor) {
+ if (tsFileProcessor == null) {
+ return CompletableFuture.completedFuture(null);
+ }
+ if (tsFileProcessor.alreadyMarkedClosing()) {
+ return tsFileProcessor.getCloseFuture();
+ }
+
+ final Set<TsFileProcessor> closingTsFileProcessors =
+ sequence ? closingSequenceTsFileProcessor : closingUnSequenceTsFileProcessor;
+ final TreeMap<Long, TsFileProcessor> workTsFileProcessors =
+ sequence ? workSequenceTsFileProcessors : workUnsequenceTsFileProcessors;
+
+ // Register before scheduling so even a very fast close callback can remove this processor.
+ closingTsFileProcessors.add(tsFileProcessor);
+ final Future<?> future = tsFileProcessor.asyncCloseForFullFlush();
+ if (!tsFileProcessor.alreadyMarkedClosing()) {
+ // An ordinary flush is still running. Keep the processor working, wait outside the locks,
+ // and retry after the flush manager has released it.
+ closingTsFileProcessors.remove(tsFileProcessor);
+ return future;
+ }
+
+ if (future != null && future.isDone()) {
+ closingTsFileProcessors.remove(tsFileProcessor);
+ }
+ final boolean removedWorkingProcessor =
+ workTsFileProcessors.get(tsFileProcessor.getTimeRangeId()) == tsFileProcessor;
+ if (removedWorkingProcessor) {
+ workTsFileProcessors.remove(tsFileProcessor.getTimeRangeId());
+ }
+
+ final TsFileResource resource = tsFileProcessor.getTsFileResource();
+ logger.info(
+ StorageEngineMessages.STORAGE_LOG_ASYNC_CLOSE_TSFILE_FILE_START_TIME_FILE_END_TIME_65020832,
+ resource.getTsFile().getAbsolutePath(),
+ resource.getFileStartTime(),
+ resource.getFileEndTime());
+ if (removedWorkingProcessor
+ && workSequenceTsFileProcessors.get(tsFileProcessor.getTimeRangeId()) == null
+ && workUnsequenceTsFileProcessors.get(tsFileProcessor.getTimeRangeId()) == null) {
+ WritingMetrics.getInstance().recordActiveTimePartitionCount(-1);
+ }
+ return future;
+ }
+
/** force close all working tsfile processors */
public void forceCloseAllWorkingTsFileProcessors() throws TsFileProcessorException {
writeLock("forceCloseAllWorkingTsFileProcessors");
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
index cc931e75fac..f233355f832 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/memtable/TsFileProcessor.java
@@ -1680,10 +1680,28 @@ public class TsFileProcessor {
/** async close one tsfile, register and close it by another thread */
public Future<?> asyncClose() {
+ return asyncClose(false);
+ }
+
+ /**
+ * Tries to close this TsFile after any previous ordinary flush has finished.
+ *
+ * <p>If an ordinary flush is still being managed by the flush manager, this method only returns
+ * that flush's future. The caller must wait for it and retry until {@link
+ * #alreadyMarkedClosing()} becomes {@code true}. This avoids adding a close signal while the old
+ * flush task is about to leave the flush manager.
+ */
+ public Future<?> asyncCloseForFullFlush() {
+ return asyncClose(true);
+ }
+
+ private Future<?> asyncClose(final boolean ignorePreviousFlushFuture) {
flushQueryLock.writeLock().lock();
logFlushQueryWriteLocked();
try {
- if (closeFuture != null) {
+ if (shouldClose
+ || managedByFlushManager
+ || (!ignorePreviousFlushFuture && closeFuture != null)) {
return closeFuture;
}
@@ -1734,12 +1752,12 @@ public class TsFileProcessor {
dataRegionName,
tsFileResource.getTsFile().getName(),
e);
+ return CompletableFuture.failedFuture(e);
}
} finally {
flushQueryLock.writeLock().unlock();
logFlushQueryWriteUnlocked();
}
- return CompletableFuture.completedFuture(null);
}
/** Put the working memtable into flushing list and set the working memtable to null */
@@ -2194,9 +2212,27 @@ public class TsFileProcessor {
}
public void setManagedByFlushManager(boolean managedByFlushManager) {
- this.managedByFlushManager = managedByFlushManager;
+ flushQueryLock.writeLock().lock();
+ try {
+ this.managedByFlushManager = managedByFlushManager;
+ if (!managedByFlushManager) {
+ closeFuture = CompletableFuture.completedFuture(null);
+ }
+ } finally {
+ flushQueryLock.writeLock().unlock();
+ }
if (!managedByFlushManager) {
- closeFuture = CompletableFuture.completedFuture(null);
+ synchronized (this) {
+ notifyAll();
+ }
+ }
+ }
+
+ public void waitUntilFlushManagerReleased() throws InterruptedException {
+ synchronized (this) {
+ while (managedByFlushManager) {
+ wait();
+ }
}
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollectorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollectorTest.java
index 029a722c8a9..de89e6c02ab 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollectorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/connection/PipeEventCollectorTest.java
@@ -53,7 +53,7 @@ public class PipeEventCollectorTest {
pendingQueue.discardEventsOfPipe("pipe", 1L, 1);
final PipeEventCollector droppedPipeCollector =
- new PipeEventCollector(pendingQueue, 1L, 1, false, false, false);
+ new PipeEventCollector(pendingQueue, 1L, 1, Long.MIN_VALUE, false, false, false);
final PipeRawTabletInsertionEvent droppedPipeEvent =
createPipeRawTabletInsertionEvent("pipe", 1L);
droppedPipeCollector.collect(droppedPipeEvent);
@@ -62,7 +62,7 @@ public class PipeEventCollectorTest {
Assert.assertEquals(0, pendingQueue.size());
final PipeEventCollector recreatedPipeCollector =
- new PipeEventCollector(pendingQueue, 2L, 1, false, false, false);
+ new PipeEventCollector(pendingQueue, 2L, 1, Long.MIN_VALUE, false, false, false);
final PipeRawTabletInsertionEvent recreatedPipeEvent =
createPipeRawTabletInsertionEvent("pipe", 2L);
recreatedPipeCollector.collect(recreatedPipeEvent);
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskTest.java
new file mode 100644
index 00000000000..cce1edd0523
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/agent/task/subtask/processor/PipeProcessorSubtaskTest.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.agent.task.subtask.processor;
+
+import org.apache.iotdb.db.pipe.agent.task.connection.PipeEventCollector;
+import org.apache.iotdb.pipe.api.PipeProcessor;
+import org.apache.iotdb.pipe.api.event.Event;
+
+import org.junit.Test;
+import org.mockito.ArgumentCaptor;
+
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
+
+public class PipeProcessorSubtaskTest {
+
+ @Test
+ public void testGenericControlEventBypassesProcessor() throws Exception {
+ final Event event =
+ new Event() {
+ @Override
+ public boolean shouldSkipProcessing() {
+ return true;
+ }
+ };
+ final PipeProcessor processor = mock(PipeProcessor.class);
+ final PipeEventCollector collector = mock(PipeEventCollector.class);
+ final PipeProcessorSubtask subtask =
+ new PipeProcessorSubtask("test", "pipe", 1L, 1, () -> event, processor, collector);
+ try {
+ assertTrue(executeOnce(subtask, collector));
+ verify(collector).collect(event);
+ verifyNoMoreInteractions(processor);
+ } finally {
+ subtask.close();
+ }
+ }
+
+ @Test
+ public void testOrdinaryEventStillReachesProcessor() throws Exception {
+ final Event event = new Event() {};
+ final PipeProcessor processor = mock(PipeProcessor.class);
+ final PipeEventCollector collector = mock(PipeEventCollector.class);
+ final PipeProcessorSubtask subtask =
+ new PipeProcessorSubtask("test", "pipe", 1L, 1, () -> event, processor, collector);
+ try {
+ assertTrue(executeOnce(subtask, collector));
+ verify(processor).process(event, collector);
+ } finally {
+ subtask.close();
+ }
+ }
+
+ private static boolean executeOnce(
+ final PipeProcessorSubtask subtask, final PipeEventCollector collector) throws Exception {
+ final ArgumentCaptor<PipeProcessorSubtaskExecutionGuard> guardCaptor =
+ ArgumentCaptor.forClass(PipeProcessorSubtaskExecutionGuard.class);
+ verify(collector).setProcessorExecutionGuard(guardCaptor.capture());
+ final PipeProcessorSubtaskExecutionGuard guard = guardCaptor.getValue();
+ subtask.onAllowSubmittingSelf();
+ guard.enter();
+ try {
+ return subtask.executeOnce();
+ } finally {
+ guard.exit();
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEventTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEventTest.java
new file mode 100644
index 00000000000..8a50fdcb0d8
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/event/common/heartbeat/PipeHeartbeatEventTest.java
@@ -0,0 +1,92 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.event.common.heartbeat;
+
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
+import org.apache.iotdb.commons.pipe.event.EnrichedEvent;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+
+public class PipeHeartbeatEventTest {
+
+ @Test
+ public void testOnlyBoundCompletionBarrierNeedsOrderedCommit() {
+ final PipeHeartbeatEvent rawBarrier = new PipeHeartbeatEvent(1, false, true);
+ rawBarrier.bindCompletionBarrier(2L, 3L);
+ assertFalse(rawBarrier.needToCommit());
+
+ final PipeTaskMeta taskMeta = mock(PipeTaskMeta.class);
+ final EnrichedEvent copiedEvent =
+ rawBarrier.shallowCopySelfAndBindPipeTaskMetaForProgressReport(
+ "test_pipe",
+ 1L,
+ taskMeta,
+ null,
+ null,
+ null,
+ null,
+ null,
+ true,
+ Long.MIN_VALUE,
+ Long.MAX_VALUE);
+ final PipeHeartbeatEvent boundBarrier = (PipeHeartbeatEvent) copiedEvent;
+
+ assertTrue(boundBarrier.isCompletionBarrier());
+ assertFalse(boundBarrier.needToCommit());
+ assertTrue(boundBarrier.getOnCommittedHooks().isEmpty());
+ boundBarrier.bindSource(4L);
+ assertTrue(boundBarrier.needToCommit());
+ assertEquals(2L, boundBarrier.getAssignerEpoch());
+ assertEquals(3L, boundBarrier.getDataGeneration());
+ assertEquals(4L, boundBarrier.getSourceId());
+ assertEquals(1, boundBarrier.getOnCommittedHooks().size());
+ boundBarrier.bindSource(5L);
+ assertEquals(4L, boundBarrier.getSourceId());
+ assertEquals(1, boundBarrier.getOnCommittedHooks().size());
+
+ final PipeHeartbeatEvent periodicHeartbeat = new PipeHeartbeatEvent(1, false);
+ final EnrichedEvent boundPeriodicHeartbeat =
+ periodicHeartbeat.shallowCopySelfAndBindPipeTaskMetaForProgressReport(
+ "test_pipe",
+ 1L,
+ taskMeta,
+ null,
+ null,
+ null,
+ null,
+ null,
+ true,
+ Long.MIN_VALUE,
+ Long.MAX_VALUE);
+ assertFalse(boundPeriodicHeartbeat.needToCommit());
+ assertTrue(boundPeriodicHeartbeat.getOnCommittedHooks().isEmpty());
+ boundPeriodicHeartbeat.bindSource(6L);
+ assertEquals(6L, boundPeriodicHeartbeat.getSourceId());
+ assertTrue(boundPeriodicHeartbeat.isControlEvent());
+ assertFalse(boundPeriodicHeartbeat.shouldSkipProcessing());
+ assertTrue(boundBarrier.shouldSkipProcessing());
+ assertTrue(boundPeriodicHeartbeat.getOnCommittedHooks().isEmpty());
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeCompletionOperatorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeCompletionOperatorTest.java
new file mode 100644
index 00000000000..cb4712d5d50
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeCompletionOperatorTest.java
@@ -0,0 +1,248 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.metric.overview;
+
+import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
+import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey;
+import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEvent;
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.PipeRealtimeDataRegionSource;
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeDataRegionAssigner;
+import org.apache.iotdb.pipe.api.event.Event;
+
+import org.apache.tsfile.utils.Pair;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.Assert.assertEquals;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class PipeDataNodeCompletionOperatorTest {
+
+ @Test
+ public void testCompletionFollowsCommittedGenerationAndFailsClosed() {
+ final AtomicBoolean remainingZero = new AtomicBoolean(true);
+ final AtomicBoolean supported = new AtomicBoolean(true);
+ final AtomicReference<Map<Integer, PipeRealtimeDataRegionSource>> expected =
+ new AtomicReference<>(new HashMap<>());
+ final AtomicReference<CommitterKey> currentCommitter = new AtomicReference<>();
+ final PipeDataNodeCompletionOperator operator =
+ new PipeDataNodeCompletionOperator(
+ remainingZero::get,
+ () -> new Pair<>(supported.get(), expected.get()),
+ key -> key == currentCommitter.get());
+
+ final PipeTaskMeta taskMeta = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 0);
+ final PipeTaskMeta staleTaskMeta = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 0);
+ final TestSource source = new TestSource(1, taskMeta);
+ final PipeDataRegionAssigner assigner = mock(PipeDataRegionAssigner.class);
+ final CommitterKey committerKey = mock(CommitterKey.class);
+ final CommitterKey nextCommitterKey = mock(CommitterKey.class);
+ final AtomicLong publishedGeneration = new AtomicLong(0);
+ final AtomicLong publicationFailureEpoch = new AtomicLong(0);
+
+ when(assigner.getAssignerEpoch()).thenReturn(10L);
+ when(assigner.getPublishedDataGeneration()).thenAnswer(i -> publishedGeneration.get());
+ when(assigner.getPublicationFailureEpoch()).thenAnswer(i -> publicationFailureEpoch.get());
+ currentCommitter.set(committerKey);
+ expected.set(Collections.singletonMap(1, source));
+
+ operator.registerDataRegionSource(source);
+ operator.register(source, assigner);
+ assertEquals(0, operator.getCompletion());
+
+ operator.markCompleted(1, staleTaskMeta, 10L, 0, source.getSourceId(), committerKey);
+ assertEquals(0, operator.getCompletion());
+
+ operator.markCompleted(1, taskMeta, 10L, 0, source.getSourceId(), committerKey);
+ assertEquals(1, operator.getCompletion());
+
+ publishedGeneration.incrementAndGet();
+ assertEquals(0, operator.getCompletion());
+ operator.markCompleted(1, taskMeta, 10L, 1, source.getSourceId(), committerKey);
+ assertEquals(1, operator.getCompletion());
+
+ source.setDegraded(true);
+ assertEquals(0, operator.getCompletion());
+ source.setDegraded(false);
+ assertEquals(1, operator.getCompletion());
+
+ remainingZero.set(false);
+ assertEquals(0, operator.getCompletion());
+ remainingZero.set(true);
+ supported.set(false);
+ assertEquals(0, operator.getCompletion());
+ supported.set(true);
+ assertEquals(1, operator.getCompletion());
+
+ currentCommitter.set(nextCommitterKey);
+ assertEquals(0, operator.getCompletion());
+ operator.markCompleted(1, taskMeta, 10L, 1, source.getSourceId(), nextCommitterKey);
+ assertEquals(1, operator.getCompletion());
+
+ publicationFailureEpoch.incrementAndGet();
+ assertEquals(0, operator.getCompletion());
+
+ operator.markInvalid(1, taskMeta, source.getSourceId());
+ operator.markCompleted(
+ 1, taskMeta, 10L, Long.MAX_VALUE, source.getSourceId(), nextCommitterKey);
+ assertEquals(0, operator.getCompletion());
+
+ operator.deregister(source, assigner);
+ assertEquals(0, operator.getCompletion());
+ }
+
+ @Test
+ public void testMembershipMustMatchExactlyAndEmptyLocalMembershipIsNeutral() {
+ final AtomicReference<Map<Integer, PipeRealtimeDataRegionSource>> expected =
+ new AtomicReference<>(new HashMap<>());
+ final PipeDataNodeCompletionOperator operator =
+ new PipeDataNodeCompletionOperator(
+ () -> true, () -> new Pair<>(true, expected.get()), key -> true);
+
+ final PipeTaskMeta firstTaskMeta = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 0);
+ final PipeTaskMeta secondTaskMeta = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 0);
+ final TestSource firstSource = new TestSource(1, firstTaskMeta);
+ final TestSource secondSource = new TestSource(2, secondTaskMeta);
+ final PipeDataRegionAssigner firstAssigner = mock(PipeDataRegionAssigner.class);
+ final PipeDataRegionAssigner secondAssigner = mock(PipeDataRegionAssigner.class);
+ final CommitterKey firstCommitter = mock(CommitterKey.class);
+ final CommitterKey secondCommitter = mock(CommitterKey.class);
+
+ when(firstAssigner.getAssignerEpoch()).thenReturn(11L);
+ when(secondAssigner.getAssignerEpoch()).thenReturn(12L);
+
+ operator.registerDataRegionSource(firstSource);
+ operator.register(firstSource, firstAssigner);
+ operator.markCompleted(1, firstTaskMeta, 11L, 0, firstSource.getSourceId(), firstCommitter);
+ expected.set(Collections.singletonMap(1, firstSource));
+ assertEquals(1, operator.getCompletion());
+
+ operator.registerDataRegionSource(secondSource);
+ operator.register(secondSource, secondAssigner);
+ operator.markCompleted(2, secondTaskMeta, 12L, 0, secondSource.getSourceId(), secondCommitter);
+ final Map<Integer, PipeRealtimeDataRegionSource> bothExpected = new HashMap<>();
+ bothExpected.put(1, firstSource);
+ bothExpected.put(2, secondSource);
+ expected.set(bothExpected);
+ assertEquals(1, operator.getCompletion());
+
+ expected.set(Collections.singletonMap(2, secondSource));
+ assertEquals(0, operator.getCompletion());
+ operator.deregister(firstSource, firstAssigner);
+ operator.deregisterDataRegionSource(firstSource);
+ assertEquals(1, operator.getCompletion());
+
+ expected.set(Collections.emptyMap());
+ assertEquals(0, operator.getCompletion());
+ operator.deregister(secondSource, secondAssigner);
+ operator.deregisterDataRegionSource(secondSource);
+ assertEquals(1, operator.getCompletion());
+ }
+
+ @Test
+ public void testStaleBarrierCannotCompleteReplacementSource() {
+ final PipeTaskMeta sharedTaskMeta = new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 0);
+ final PipeDataRegionAssigner sharedAssigner = mock(PipeDataRegionAssigner.class);
+ final CommitterKey committerKey = mock(CommitterKey.class);
+ when(sharedAssigner.getAssignerEpoch()).thenReturn(20L);
+
+ final AtomicReference<Map<Integer, PipeRealtimeDataRegionSource>> expected =
+ new AtomicReference<>(new HashMap<>());
+ final PipeDataNodeCompletionOperator operator =
+ new PipeDataNodeCompletionOperator(
+ () -> true, () -> new Pair<>(true, expected.get()), key -> true);
+
+ final TestSource oldSource = new TestSource(1, sharedTaskMeta);
+ operator.registerDataRegionSource(oldSource);
+ operator.register(oldSource, sharedAssigner);
+ operator.deregister(oldSource, sharedAssigner);
+ operator.deregisterDataRegionSource(oldSource);
+
+ final TestSource newSource = new TestSource(1, sharedTaskMeta);
+ expected.set(Collections.singletonMap(1, newSource));
+ operator.registerDataRegionSource(newSource);
+ operator.register(newSource, sharedAssigner);
+
+ operator.markCompleted(1, sharedTaskMeta, 20L, 0, oldSource.getSourceId(), committerKey);
+ assertEquals(0, operator.getCompletion());
+
+ operator.markCompleted(1, sharedTaskMeta, 20L, 0, newSource.getSourceId(), committerKey);
+ assertEquals(1, operator.getCompletion());
+
+ operator.markInvalid(1, sharedTaskMeta, oldSource.getSourceId());
+ assertEquals(1, operator.getCompletion());
+
+ operator.deregister(oldSource, sharedAssigner);
+ operator.deregisterDataRegionSource(oldSource);
+ assertEquals(1, operator.getCompletion());
+
+ operator.markInvalid(1, sharedTaskMeta, newSource.getSourceId());
+ assertEquals(0, operator.getCompletion());
+ }
+
+ private static class TestSource extends PipeRealtimeDataRegionSource {
+
+ private final AtomicBoolean degraded = new AtomicBoolean(false);
+
+ private TestSource(final int dataRegionId, final PipeTaskMeta pipeTaskMeta) {
+ this.dataRegionId = dataRegionId;
+ this.pipeTaskMeta = pipeTaskMeta;
+ }
+
+ private void setDegraded(final boolean degraded) {
+ if (this.degraded.getAndSet(degraded) != degraded) {
+ markCompletionStateChanged();
+ }
+ }
+
+ @Override
+ protected void doExtract(final PipeRealtimeEvent event) {
+ // Do nothing.
+ }
+
+ @Override
+ public Event supply() {
+ return null;
+ }
+
+ @Override
+ public boolean isTsFileEpochDegraded() {
+ return degraded.get();
+ }
+
+ @Override
+ public boolean isNeedListenToTsFile() {
+ return false;
+ }
+
+ @Override
+ public boolean isNeedListenToInsertNode() {
+ return false;
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeSinglePipeMetricsTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeSinglePipeMetricsTest.java
new file mode 100644
index 00000000000..8246f9080d3
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/metric/overview/PipeDataNodeSinglePipeMetricsTest.java
@@ -0,0 +1,221 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.metric.overview;
+
+import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
+import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
+import org.apache.iotdb.commons.pipe.agent.task.progress.CommitterKey;
+import org.apache.iotdb.commons.service.metric.enums.Metric;
+import org.apache.iotdb.commons.service.metric.enums.Tag;
+import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEvent;
+import org.apache.iotdb.db.pipe.source.dataregion.IoTDBDataRegionSource;
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.PipeRealtimeDataRegionSource;
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeDataRegionAssigner;
+import org.apache.iotdb.metrics.AbstractMetricService;
+import org.apache.iotdb.metrics.utils.MetricLevel;
+import org.apache.iotdb.metrics.utils.MetricType;
+import org.apache.iotdb.pipe.api.event.Event;
+
+import org.junit.Test;
+import org.mockito.invocation.Invocation;
+
+import java.lang.reflect.Field;
+import java.util.Arrays;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockingDetails;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class PipeDataNodeSinglePipeMetricsTest {
+
+ @Test
+ public void testCompletionGaugeLifecycleAndStaleCallbacksDoNotRecreateState() throws Exception {
+ final String pipeName = "completion_metric_" + System.nanoTime();
+ final long creationTime = 1L;
+ final String pipeId = pipeName + "_" + creationTime;
+ final TestSource realtimeSource =
+ new TestSource(
+ pipeName, creationTime, 1, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 0));
+ final IoTDBDataRegionSource outerSource = mock(IoTDBDataRegionSource.class);
+ final PipeDataRegionAssigner assigner = mock(PipeDataRegionAssigner.class);
+ final AbstractMetricService metricService = mock(AbstractMetricService.class);
+
+ when(outerSource.getPipeName()).thenReturn(pipeName);
+ when(outerSource.getCreationTime()).thenReturn(creationTime);
+ when(outerSource.getRealtimeSourceForCompletion()).thenReturn(realtimeSource);
+
+ final PipeDataNodeSinglePipeMetrics metrics = PipeDataNodeSinglePipeMetrics.getInstance();
+ final Field metricServiceField =
+ PipeDataNodeSinglePipeMetrics.class.getDeclaredField("metricService");
+ metricServiceField.setAccessible(true);
+ reset(metrics, metricServiceField);
+
+ try {
+ metrics.register(outerSource);
+ metrics.register(realtimeSource, assigner);
+ metrics.bindTo(metricService);
+
+ assertTrue(hasCompletionGaugeCreation(metricService, pipeName, creationTime));
+
+ metrics.deregister(pipeId);
+ assertFalse(metrics.remainingEventAndTimeOperatorMap.containsKey(pipeId));
+ assertFalse(metrics.completionOperatorMap.containsKey(pipeId));
+ verify(metricService)
+ .remove(
+ MetricType.AUTO_GAUGE,
+ Metric.PIPE_DATANODE_COMPLETION_READY.toString(),
+ Tag.NAME.toString(),
+ pipeName,
+ Tag.CREATION_TIME.toString(),
+ String.valueOf(creationTime));
+
+ metrics.register(realtimeSource, assigner);
+ metrics.markDataRegionCompleted(
+ pipeName,
+ creationTime,
+ 1,
+ realtimeSource.getPipeTaskMeta(),
+ 1,
+ 1,
+ realtimeSource.getSourceId(),
+ mock(CommitterKey.class));
+ metrics.markDataRegionCompletionInvalid(
+ pipeName,
+ creationTime,
+ 1,
+ realtimeSource.getPipeTaskMeta(),
+ realtimeSource.getSourceId());
+ metrics.deregister(realtimeSource, assigner);
+ metrics.decreaseHeartbeatEventCount(pipeName, creationTime);
+ assertFalse(metrics.remainingEventAndTimeOperatorMap.containsKey(pipeId));
+ assertFalse(metrics.completionOperatorMap.containsKey(pipeId));
+ } finally {
+ reset(metrics, metricServiceField);
+ }
+ }
+
+ @Test
+ public void testDeregisterClearsOperatorsWhenMetricsAreUnbound() throws Exception {
+ final String pipeName = "completion_unbound_" + System.nanoTime();
+ final long creationTime = 2L;
+ final String pipeId = pipeName + "_" + creationTime;
+ final TestSource realtimeSource =
+ new TestSource(
+ pipeName, creationTime, 2, new PipeTaskMeta(MinimumProgressIndex.INSTANCE, 0));
+ final IoTDBDataRegionSource outerSource = mock(IoTDBDataRegionSource.class);
+ when(outerSource.getPipeName()).thenReturn(pipeName);
+ when(outerSource.getCreationTime()).thenReturn(creationTime);
+ when(outerSource.getRealtimeSourceForCompletion()).thenReturn(realtimeSource);
+ when(outerSource.getHistoricalTsFileInsertionEventCount()).thenReturn(1);
+
+ final PipeDataNodeSinglePipeMetrics metrics = PipeDataNodeSinglePipeMetrics.getInstance();
+ final Field metricServiceField =
+ PipeDataNodeSinglePipeMetrics.class.getDeclaredField("metricService");
+ metricServiceField.setAccessible(true);
+ reset(metrics, metricServiceField);
+
+ try {
+ metrics.register(outerSource);
+ assertTrue(metrics.remainingEventAndTimeOperatorMap.containsKey(pipeId));
+ assertTrue(metrics.completionOperatorMap.containsKey(pipeId));
+ assertEquals(
+ 1, metrics.remainingEventAndTimeOperatorMap.get(pipeId).getRemainingNonHeartbeatEvents());
+
+ metrics.deregister(outerSource);
+ assertEquals(
+ 0, metrics.remainingEventAndTimeOperatorMap.get(pipeId).getRemainingNonHeartbeatEvents());
+
+ metrics.deregister(pipeId);
+ assertFalse(metrics.remainingEventAndTimeOperatorMap.containsKey(pipeId));
+ assertFalse(metrics.completionOperatorMap.containsKey(pipeId));
+ } finally {
+ reset(metrics, metricServiceField);
+ }
+ }
+
+ private static boolean hasCompletionGaugeCreation(
+ final AbstractMetricService metricService, final String pipeName, final long creationTime) {
+ for (final Invocation invocation : mockingDetails(metricService).getInvocations()) {
+ final Object[] arguments = invocation.getRawArguments();
+ if ("createAutoGauge".equals(invocation.getMethod().getName())
+ && arguments.length == 5
+ && Metric.PIPE_DATANODE_COMPLETION_READY.toString().equals(arguments[0])
+ && MetricLevel.IMPORTANT.equals(arguments[1])
+ && arguments[2] instanceof PipeDataNodeCompletionOperator
+ && Arrays.equals(
+ new String[] {
+ Tag.NAME.toString(),
+ pipeName,
+ Tag.CREATION_TIME.toString(),
+ String.valueOf(creationTime)
+ },
+ (String[]) arguments[4])) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static void reset(
+ final PipeDataNodeSinglePipeMetrics metrics, final Field metricServiceField)
+ throws IllegalAccessException {
+ metrics.remainingEventAndTimeOperatorMap.clear();
+ metrics.completionOperatorMap.clear();
+ metricServiceField.set(metrics, null);
+ }
+
+ private static class TestSource extends PipeRealtimeDataRegionSource {
+
+ private TestSource(
+ final String pipeName,
+ final long creationTime,
+ final int dataRegionId,
+ final PipeTaskMeta pipeTaskMeta) {
+ this.pipeName = pipeName;
+ this.creationTime = creationTime;
+ this.dataRegionId = dataRegionId;
+ this.pipeTaskMeta = pipeTaskMeta;
+ }
+
+ @Override
+ protected void doExtract(final PipeRealtimeEvent event) {
+ // Do nothing.
+ }
+
+ @Override
+ public Event supply() {
+ return null;
+ }
+
+ @Override
+ public boolean isNeedListenToTsFile() {
+ return false;
+ }
+
+ @Override
+ public boolean isNeedListenToInsertNode() {
+ return false;
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSinkTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSinkTest.java
index 1e6f3f865c1..ca88f30fcc5 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSinkTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/sink/protocol/airgap/IoTDBDataRegionAirGapSinkTest.java
@@ -30,6 +30,7 @@ import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTable
import org.apache.iotdb.db.pipe.sink.payload.evolvable.request.PipeTransferTsFilePieceReq;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameterValidator;
import org.apache.iotdb.pipe.api.customizer.parameter.PipeParameters;
+import org.apache.iotdb.pipe.api.exception.PipeConnectionException;
import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
import org.apache.tsfile.enums.TSDataType;
@@ -40,6 +41,7 @@ import org.apache.tsfile.write.schema.MeasurementSchema;
import org.junit.Assert;
import org.junit.Test;
+import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
@@ -50,6 +52,31 @@ import java.util.concurrent.atomic.AtomicLong;
public class IoTDBDataRegionAirGapSinkTest {
+ @Test
+ public void testCompletionBarrierRetriesUnacknowledgedBatch() throws Exception {
+ try (final RecordingIoTDBDataRegionAirGapSink sink = new RecordingIoTDBDataRegionAirGapSink()) {
+ final PipeParameters parameters = buildParameters(false);
+ parameters.getAttribute().put(PipeSinkConstant.CONNECTOR_IOTDB_BATCH_SIZE_KEY, "1");
+ sink.validate(new PipeParameterValidator(parameters));
+ sink.customize(
+ parameters,
+ new PipeTaskRuntimeConfiguration(new PipeTaskSinkRuntimeEnvironment("pipe", 1L, 1)));
+ sink.prepareSocket();
+
+ final PipeRawTabletInsertionEvent tablet = createPipeRawTabletInsertionEvent("pipe", 1L, 1L);
+ sink.failNextSend = true;
+ Assert.assertThrows(PipeConnectionException.class, () -> sink.transfer(tablet));
+ Assert.assertFalse(tablet.isReleased());
+ Assert.assertEquals(1, sink.getBatchSize());
+
+ sink.transfer(new PipeHeartbeatEvent(1, false, true));
+ Assert.assertEquals(0, sink.getBatchSize());
+ Assert.assertTrue(tablet.isReleased());
+ Assert.assertEquals(2, sink.sentRequests.size());
+ Assert.assertArrayEquals(sink.sentRequests.get(0), sink.sentRequests.get(1));
+ }
+ }
+
@Test
public void testTransferTabletBatchOverAirGap() throws Exception {
try (final RecordingIoTDBDataRegionAirGapSink sink = new RecordingIoTDBDataRegionAirGapSink()) {
@@ -162,6 +189,7 @@ public class IoTDBDataRegionAirGapSinkTest {
private final List<byte[]> sentRequests = new ArrayList<>();
private final AtomicLong rateLimitedBytes = new AtomicLong(0);
+ private boolean failNextSend;
private void prepareSocket() {
sockets.set(0, new TestingAirGapSocket());
@@ -173,8 +201,12 @@ public class IoTDBDataRegionAirGapSinkTest {
}
@Override
- protected boolean sendBytes(final AirGapSocket socket, final byte[] bytes) {
+ protected boolean sendBytes(final AirGapSocket socket, final byte[] bytes) throws IOException {
sentRequests.add(Arrays.copyOf(bytes, bytes.length));
+ if (failNextSend) {
+ failNextSend = false;
+ throw new IOException("Receiver ACK was lost");
+ }
return true;
}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSourceTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSourceTest.java
index 4677f2126b6..d935821cdba 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSourceTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/PipeRealtimeDataRegionSourceTest.java
@@ -68,6 +68,25 @@ public class PipeRealtimeDataRegionSourceTest {
}
}
+ @Test
+ public void completionBarrierIsNeverDroppedByProgressOrHeartbeatMerging() throws Exception {
+ try (final ProgressReportTestSource source = new ProgressReportTestSource()) {
+ final PipeRealtimeEvent heartbeatEvent = createHeartbeatEvent();
+ final PipeRealtimeEvent completionBarrier = createCompletionBarrierEvent();
+ source.extract(heartbeatEvent);
+ source.extract(completionBarrier);
+
+ Assert.assertFalse(heartbeatEvent.getEvent().isReleased());
+ Assert.assertFalse(completionBarrier.getEvent().isReleased());
+ Assert.assertEquals(2, source.getEventCount());
+
+ final PipeRealtimeEvent progressReportEvent = createProgressReportEvent();
+ source.extract(progressReportEvent);
+ Assert.assertFalse(completionBarrier.getEvent().isReleased());
+ Assert.assertEquals(3, source.getEventCount());
+ }
+ }
+
private static PipeRealtimeEvent createHeartbeatEvent() {
final PipeRealtimeEvent event = PipeRealtimeEventFactory.createRealtimeEvent(1, false);
Assert.assertTrue(event.increaseReferenceCount(TEST_REFERENCE_HOLDER));
@@ -84,6 +103,12 @@ public class PipeRealtimeDataRegionSourceTest {
return event;
}
+ private static PipeRealtimeEvent createCompletionBarrierEvent() {
+ final PipeRealtimeEvent event = PipeRealtimeEventFactory.createCompletionBarrierEvent(1);
+ Assert.assertTrue(event.increaseReferenceCount(TEST_REFERENCE_HOLDER));
+ return event;
+ }
+
private static class ProgressReportTestSource extends PipeRealtimeDataRegionSource {
@Override
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssignerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssignerTest.java
new file mode 100644
index 00000000000..bfc8618f1f1
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/assigner/PipeDataRegionAssignerTest.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner;
+
+import org.apache.iotdb.commons.pipe.event.ProgressReportEvent;
+import org.apache.iotdb.db.pipe.event.realtime.PipeRealtimeEventFactory;
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeDataRegionAssigner.CompletionToken;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class PipeDataRegionAssignerTest {
+
+ @Test
+ public void testNonHeartbeatControlEventDoesNotAdvanceDataGeneration() {
+ try (final PipeDataRegionAssigner assigner = new PipeDataRegionAssigner(Integer.MIN_VALUE)) {
+ final CompletionToken token = assigner.invalidateCompletionAndGetToken();
+ final long generation = assigner.getPublishedDataGeneration();
+ assigner.publishToAssign(
+ PipeRealtimeEventFactory.createRealtimeEvent(new ProgressReportEvent(null, 0L, null)));
+ assertEquals(generation, assigner.getPublishedDataGeneration());
+ assertTrue(assigner.publishCompletionBarrier(token));
+ }
+ }
+
+ @Test
+ public void testNullDataEventMarksPublicationFailed() {
+ try (final PipeDataRegionAssigner assigner = new PipeDataRegionAssigner(Integer.MIN_VALUE)) {
+ final long generation = assigner.getPublishedDataGeneration();
+ final long failureEpoch = assigner.getPublicationFailureEpoch();
+
+ assigner.publishDataEventToAssign(() -> null);
+
+ assertEquals(generation + 1, assigner.getPublishedDataGeneration());
+ assertEquals(failureEpoch + 1, assigner.getPublicationFailureEpoch());
+ }
+ }
+
+ @Test
+ public void testOnlyLatestCompletionInvalidationCanPublishBarrier() {
+ try (final PipeDataRegionAssigner assigner = new PipeDataRegionAssigner(Integer.MIN_VALUE)) {
+ final CompletionToken staleToken = assigner.invalidateCompletionAndGetToken();
+ final CompletionToken currentToken = assigner.invalidateCompletionAndGetToken();
+
+ assertFalse(assigner.publishCompletionBarrier(staleToken));
+ assigner.invalidateCompletion();
+ assertTrue(assigner.publishCompletionBarrier(currentToken));
+ }
+ }
+
+ @Test
+ public void testInsertInvalidationRejectsCurrentCompletionToken() {
+ try (final PipeDataRegionAssigner assigner = new PipeDataRegionAssigner(Integer.MIN_VALUE)) {
+ final long generation = assigner.getPublishedDataGeneration();
+ final CompletionToken ignoredInsertToken = assigner.invalidateCompletionAndGetToken();
+
+ assigner.invalidateCompletionBarrier();
+
+ assertEquals(generation + 2, assigner.getPublishedDataGeneration());
+ assertFalse(assigner.publishCompletionBarrier(ignoredInsertToken));
+
+ final CompletionToken publishedInsertToken = assigner.invalidateCompletionAndGetToken();
+ assigner.publishInsertDataEventToAssign(
+ () -> PipeRealtimeEventFactory.createRealtimeEvent(Integer.MIN_VALUE, false));
+ assertFalse(assigner.publishCompletionBarrier(publishedInsertToken));
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeInsertionDataNodeListenerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeInsertionDataNodeListenerTest.java
new file mode 100644
index 00000000000..08240ae24f7
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/pipe/source/dataregion/realtime/listener/PipeInsertionDataNodeListenerTest.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.pipe.source.dataregion.realtime.listener;
+
+import org.apache.iotdb.db.pipe.source.dataregion.realtime.assigner.PipeDataRegionAssigner;
+
+import org.junit.Test;
+
+import java.lang.reflect.Field;
+import java.util.concurrent.ConcurrentMap;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class PipeInsertionDataNodeListenerTest {
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testInsertUsesBarrierInvalidatingPublicationPath() throws Exception {
+ final PipeInsertionDataNodeListener listener = PipeInsertionDataNodeListener.getInstance();
+ final Field assignerMapField =
+ PipeInsertionDataNodeListener.class.getDeclaredField("dataRegionId2Assigner");
+ assignerMapField.setAccessible(true);
+ final ConcurrentMap<Integer, PipeDataRegionAssigner> assignerMap =
+ (ConcurrentMap<Integer, PipeDataRegionAssigner>) assignerMapField.get(listener);
+ final int dataRegionId = Integer.MIN_VALUE + 1;
+ final PipeDataRegionAssigner assigner = mock(PipeDataRegionAssigner.class);
+ assignerMap.put(dataRegionId, assigner);
+
+ try {
+ when(assigner.shouldListenToInsertNode()).thenReturn(false);
+ listener.listenToInsertNode(dataRegionId, null, null, null);
+ verify(assigner).invalidateCompletionBarrier();
+ verify(assigner, never()).invalidateCompletion();
+
+ reset(assigner);
+ when(assigner.shouldListenToInsertNode()).thenReturn(true);
+ listener.listenToInsertNode(dataRegionId, null, null, null);
+ verify(assigner).publishInsertDataEventToAssign(any());
+ verify(assigner, never()).publishDataEventToAssign(any());
+ } finally {
+ assignerMap.remove(dataRegionId, assigner);
+ }
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java
index b287a669b80..493f4971ed6 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/DataRegionTest.java
@@ -65,8 +65,10 @@ import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.Compacti
import org.apache.iotdb.db.storageengine.dataregion.compaction.selector.constant.InnerSequenceCompactionSelector;
import org.apache.iotdb.db.storageengine.dataregion.compaction.selector.constant.InnerUnsequenceCompactionSelector;
import org.apache.iotdb.db.storageengine.dataregion.compaction.utils.CompactionConfigRestorer;
+import org.apache.iotdb.db.storageengine.dataregion.flush.FlushListener;
import org.apache.iotdb.db.storageengine.dataregion.flush.FlushManager;
import org.apache.iotdb.db.storageengine.dataregion.flush.TsFileFlushPolicy;
+import org.apache.iotdb.db.storageengine.dataregion.memtable.IMemTable;
import org.apache.iotdb.db.storageengine.dataregion.memtable.ReadOnlyMemChunk;
import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessor;
import org.apache.iotdb.db.storageengine.dataregion.modification.DeletionPredicate;
@@ -117,12 +119,14 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import static org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils.genInsertRowNode;
@@ -2419,6 +2423,143 @@ public class DataRegionTest {
assertTrue(tsFileResourceSeq.isClosed());
}
+ @Test
+ public void testFullFlushClosesProcessorAfterOrdinaryAsyncFlush() throws Exception {
+ final TSRecord record = new TSRecord(deviceId, 100);
+ record.addTuple(DataPoint.getDataPoint(TSDataType.INT32, measurementId, String.valueOf(100)));
+ dataRegion.insert(buildInsertRowNodeByTSRecord(record));
+
+ final TsFileProcessor tsFileProcessor =
+ dataRegion.getWorkSequenceTsFileProcessors().iterator().next();
+ final TsFileResource tsFileResource = tsFileProcessor.getTsFileResource();
+ tsFileProcessor.asyncFlush();
+ final Future<?> ordinaryFlushFuture = tsFileProcessor.getCloseFuture();
+ Assert.assertNotNull(ordinaryFlushFuture);
+ ordinaryFlushFuture.get();
+
+ Assert.assertFalse(tsFileProcessor.alreadyMarkedClosing());
+ Assert.assertFalse(tsFileResource.isClosed());
+
+ dataRegion.syncCloseAllWorkingAndClosingTsFileProcessors();
+
+ Assert.assertTrue(tsFileResource.isClosed());
+ Assert.assertFalse(dataRegion.getWorkSequenceTsFileProcessors().contains(tsFileProcessor));
+ }
+
+ @Test
+ public void testFullFlushWaitsForRunningOrdinaryAsyncFlush() throws Exception {
+ final CountDownLatch ordinaryFlushBlocked = new CountDownLatch(1);
+ final CountDownLatch allowOrdinaryFlush = new CountDownLatch(1);
+ dataRegion.setCustomFlushListeners(
+ Collections.singletonList(
+ new FlushListener() {
+ @Override
+ public void onMemTableFlushStarted(final IMemTable memTable) {
+ // Do nothing.
+ }
+
+ @Override
+ public void onMemTableFlushed(final IMemTable memTable) {
+ ordinaryFlushBlocked.countDown();
+ try {
+ allowOrdinaryFlush.await();
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new AssertionError(e);
+ }
+ }
+ }));
+
+ final TSRecord record = new TSRecord(deviceId, 100);
+ record.addTuple(DataPoint.getDataPoint(TSDataType.INT32, measurementId, String.valueOf(100)));
+ dataRegion.insert(buildInsertRowNodeByTSRecord(record));
+ final TsFileProcessor tsFileProcessor =
+ dataRegion.getWorkSequenceTsFileProcessors().iterator().next();
+ final TsFileResource tsFileResource = tsFileProcessor.getTsFileResource();
+
+ tsFileProcessor.asyncFlush();
+ Assert.assertTrue(ordinaryFlushBlocked.await(30, TimeUnit.SECONDS));
+ final CompletableFuture<Void> fullFlushFuture =
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ dataRegion.syncCloseAllWorkingAndClosingTsFileProcessors();
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new CompletionException(e);
+ } catch (final ExecutionException e) {
+ throw new CompletionException(e);
+ }
+ });
+
+ try {
+ fullFlushFuture.get(200, TimeUnit.MILLISECONDS);
+ Assert.fail();
+ } catch (final TimeoutException expected) {
+ // The full flush must wait until the ordinary flush releases the processor.
+ } finally {
+ allowOrdinaryFlush.countDown();
+ }
+
+ fullFlushFuture.get(30, TimeUnit.SECONDS);
+ Assert.assertTrue(tsFileResource.isClosed());
+ Assert.assertFalse(dataRegion.getWorkSequenceTsFileProcessors().contains(tsFileProcessor));
+ }
+
+ @Test
+ public void testFullFlushDoesNotChaseProcessorCreatedAfterSnapshot() throws Exception {
+ final CountDownLatch closeListenerBlocked = new CountDownLatch(1);
+ final CountDownLatch allowCloseListener = new CountDownLatch(1);
+ dataRegion.setCustomCloseFileListeners(
+ Collections.singletonList(
+ processor -> {
+ closeListenerBlocked.countDown();
+ try {
+ allowCloseListener.await();
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new TsFileProcessorException(e);
+ }
+ }));
+
+ final TSRecord firstRecord = new TSRecord(deviceId, 100);
+ firstRecord.addTuple(
+ DataPoint.getDataPoint(TSDataType.INT32, measurementId, String.valueOf(100)));
+ dataRegion.insert(buildInsertRowNodeByTSRecord(firstRecord));
+ final TsFileProcessor initialProcessor =
+ dataRegion.getWorkSequenceTsFileProcessors().iterator().next();
+
+ final CompletableFuture<Void> fullFlushFuture =
+ CompletableFuture.runAsync(
+ () -> {
+ try {
+ dataRegion.syncCloseAllWorkingAndClosingTsFileProcessors();
+ } catch (final InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new CompletionException(e);
+ } catch (final ExecutionException e) {
+ throw new CompletionException(e);
+ }
+ });
+
+ TsFileProcessor replacementProcessor;
+ try {
+ Assert.assertTrue(closeListenerBlocked.await(30, TimeUnit.SECONDS));
+ final TSRecord secondRecord = new TSRecord(deviceId, 101);
+ secondRecord.addTuple(
+ DataPoint.getDataPoint(TSDataType.INT32, measurementId, String.valueOf(101)));
+ dataRegion.insert(buildInsertRowNodeByTSRecord(secondRecord));
+ replacementProcessor = dataRegion.getWorkSequenceTsFileProcessors().iterator().next();
+ Assert.assertNotSame(initialProcessor, replacementProcessor);
+ } finally {
+ allowCloseListener.countDown();
+ }
+
+ fullFlushFuture.get(30, TimeUnit.SECONDS);
+ Assert.assertTrue(dataRegion.getWorkSequenceTsFileProcessors().contains(replacementProcessor));
+ Assert.assertFalse(replacementProcessor.alreadyMarkedClosing());
+ }
+
@Test
@SuppressWarnings("unchecked")
public void testSyncCloseWaitsForAlreadyClosingProcessor() throws Exception {
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/BuiltinPipePlugin.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/BuiltinPipePlugin.java
index 4f3307d1544..96ab486ebd7 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/BuiltinPipePlugin.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/plugin/builtin/BuiltinPipePlugin.java
@@ -53,13 +53,13 @@ public enum BuiltinPipePlugin {
// extractors
DO_NOTHING_EXTRACTOR("do-nothing-extractor", DoNothingSource.class),
- IOTDB_EXTRACTOR("iotdb-extractor", IoTDBSource.class),
+ IOTDB_EXTRACTOR("iotdb-extractor", IoTDBSource.class, true),
DO_NOTHING_SOURCE("do-nothing-source", DoNothingSource.class),
- IOTDB_SOURCE("iotdb-source", IoTDBSource.class),
+ IOTDB_SOURCE("iotdb-source", IoTDBSource.class, true),
// processors
- DO_NOTHING_PROCESSOR("do-nothing-processor", DoNothingProcessor.class),
+ DO_NOTHING_PROCESSOR("do-nothing-processor", DoNothingProcessor.class, true),
TUMBLING_TIME_SAMPLING_PROCESSOR(
"tumbling-time-sampling-processor", TumblingTimeSamplingProcessor.class),
SDT_SAMPLING_PROCESSOR("sdt-sampling-processor", SwingingDoorTrendingSamplingProcessor.class),
@@ -79,12 +79,12 @@ public enum BuiltinPipePlugin {
// connectors
DO_NOTHING_CONNECTOR("do-nothing-connector", DoNothingSink.class),
- IOTDB_THRIFT_CONNECTOR("iotdb-thrift-connector", IoTDBThriftSink.class),
- IOTDB_THRIFT_SSL_CONNECTOR("iotdb-thrift-ssl-connector", IoTDBThriftSslSink.class),
- IOTDB_THRIFT_SYNC_CONNECTOR("iotdb-thrift-sync-connector", IoTDBThriftSyncSink.class),
- IOTDB_THRIFT_ASYNC_CONNECTOR("iotdb-thrift-async-connector", IoTDBThriftAsyncSink.class),
+ IOTDB_THRIFT_CONNECTOR("iotdb-thrift-connector", IoTDBThriftSink.class, true),
+ IOTDB_THRIFT_SSL_CONNECTOR("iotdb-thrift-ssl-connector", IoTDBThriftSslSink.class, true),
+ IOTDB_THRIFT_SYNC_CONNECTOR("iotdb-thrift-sync-connector", IoTDBThriftSyncSink.class, true),
+ IOTDB_THRIFT_ASYNC_CONNECTOR("iotdb-thrift-async-connector", IoTDBThriftAsyncSink.class, true),
IOTDB_LEGACY_PIPE_CONNECTOR("iotdb-legacy-pipe-connector", IoTDBLegacyPipeSink.class),
- IOTDB_AIR_GAP_CONNECTOR("iotdb-air-gap-connector", IoTDBAirGapSink.class),
+ IOTDB_AIR_GAP_CONNECTOR("iotdb-air-gap-connector", IoTDBAirGapSink.class, true),
IOT_CONSENSUS_V2_ASYNC_CONNECTOR(
"iot-consensus-v2-async-connector", IoTConsensusV2AsyncSink.class),
// Legacy alias for stale PipeMeta written before the PipeConsensus -> IoTConsensusV2 rename.
@@ -95,12 +95,12 @@ public enum BuiltinPipePlugin {
WRITE_BACK_CONNECTOR("write-back-connector", WriteBackSink.class),
DO_NOTHING_SINK("do-nothing-sink", DoNothingSink.class),
- IOTDB_THRIFT_SINK("iotdb-thrift-sink", IoTDBThriftSink.class),
- IOTDB_THRIFT_SSL_SINK("iotdb-thrift-ssl-sink", IoTDBThriftSslSink.class),
- IOTDB_THRIFT_SYNC_SINK("iotdb-thrift-sync-sink", IoTDBThriftSyncSink.class),
- IOTDB_THRIFT_ASYNC_SINK("iotdb-thrift-async-sink", IoTDBThriftAsyncSink.class),
+ IOTDB_THRIFT_SINK("iotdb-thrift-sink", IoTDBThriftSink.class, true),
+ IOTDB_THRIFT_SSL_SINK("iotdb-thrift-ssl-sink", IoTDBThriftSslSink.class, true),
+ IOTDB_THRIFT_SYNC_SINK("iotdb-thrift-sync-sink", IoTDBThriftSyncSink.class, true),
+ IOTDB_THRIFT_ASYNC_SINK("iotdb-thrift-async-sink", IoTDBThriftAsyncSink.class, true),
IOTDB_LEGACY_PIPE_SINK("iotdb-legacy-pipe-sink", IoTDBLegacyPipeSink.class),
- IOTDB_AIR_GAP_SINK("iotdb-air-gap-sink", IoTDBAirGapSink.class),
+ IOTDB_AIR_GAP_SINK("iotdb-air-gap-sink", IoTDBAirGapSink.class, true),
WEBSOCKET_SINK("websocket-sink", WebSocketSink.class),
OPC_DA_SINK("opc-da-sink", OpcDaSink.class),
WRITE_BACK_SINK("write-back-sink", WriteBackSink.class),
@@ -113,11 +113,31 @@ public enum BuiltinPipePlugin {
private final String pipePluginName;
private final Class<?> pipePluginClass;
private final String className;
+ // The plugin preserves the ordered completion protocol. Sink support still requires sync load.
+ private final boolean supportCompletionMark;
BuiltinPipePlugin(String pipePluginName, Class<?> pipePluginClass) {
+ this(pipePluginName, pipePluginClass, false);
+ }
+
+ BuiltinPipePlugin(
+ String pipePluginName, Class<?> pipePluginClass, boolean supportCompletionMark) {
this.pipePluginName = pipePluginName;
this.pipePluginClass = pipePluginClass;
this.className = pipePluginClass.getName();
+ this.supportCompletionMark = supportCompletionMark;
+ }
+
+ public boolean supportsCompletionMark() {
+ return supportCompletionMark;
+ }
+
+ public static boolean supportsCompletionMark(final String pipePluginName) {
+ return Arrays.stream(values())
+ .anyMatch(
+ plugin ->
+ plugin.supportCompletionMark
+ && plugin.pipePluginName.equalsIgnoreCase(pipePluginName));
}
public String getPipePluginName() {
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeStaticMeta.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeStaticMeta.java
index f798266e7d2..da215ae4254 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeStaticMeta.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeStaticMeta.java
@@ -21,6 +21,7 @@ package org.apache.iotdb.commons.pipe.agent.task.meta;
import org.apache.iotdb.commons.conf.IoTDBConstant;
import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
+import org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant;
import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant;
import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
import org.apache.iotdb.commons.pipe.datastructure.visibility.Visibility;
@@ -49,6 +50,10 @@ public class PipeStaticMeta {
private PipeParameters processorParameters;
private PipeParameters sinkParameters;
+ // Derived from static plugin/load parameters once per metadata incarnation; never serialized.
+ // Runtime health, task membership, and extraction state must still be checked on every sample.
+ private boolean supportCompletionMark;
+
private PipeStaticMeta() {
// Empty constructor
}
@@ -64,6 +69,32 @@ public class PipeStaticMeta {
sourceParameters = new PipeParameters(extractorAttributes);
processorParameters = new PipeParameters(processorAttributes);
sinkParameters = new PipeParameters(connectorAttributes);
+ updateCompletionSupport();
+ }
+
+ public boolean supportsCompletionMark() {
+ return supportCompletionMark;
+ }
+
+ private void updateCompletionSupport() {
+ supportCompletionMark =
+ getPipeType() == PipeType.USER
+ && BuiltinPipePlugin.supportsCompletionMark(
+ sourceParameters.getStringOrDefault(
+ Arrays.asList(PipeSourceConstant.EXTRACTOR_KEY, PipeSourceConstant.SOURCE_KEY),
+ BuiltinPipePlugin.IOTDB_EXTRACTOR.getPipePluginName()))
+ && BuiltinPipePlugin.supportsCompletionMark(
+ processorParameters.getStringOrDefault(
+ PipeProcessorConstant.PROCESSOR_KEY,
+ BuiltinPipePlugin.DO_NOTHING_PROCESSOR.getPipePluginName()))
+ && BuiltinPipePlugin.supportsCompletionMark(
+ PipeSinkConstant.getConnectorOrSinkNameWithDefault(sinkParameters))
+ && PipeSinkConstant.CONNECTOR_LOAD_TSFILE_STRATEGY_SYNC_VALUE.equalsIgnoreCase(
+ sinkParameters.getStringOrDefault(
+ Arrays.asList(
+ PipeSinkConstant.CONNECTOR_LOAD_TSFILE_STRATEGY_KEY,
+ PipeSinkConstant.SINK_LOAD_TSFILE_STRATEGY_KEY),
+ PipeSinkConstant.CONNECTOR_LOAD_TSFILE_STRATEGY_SYNC_VALUE));
}
public String getPipeName() {
@@ -214,6 +245,7 @@ public class PipeStaticMeta {
pipeStaticMeta.sinkParameters.getAttribute().put(key, value);
}
+ pipeStaticMeta.updateCompletionSupport();
return pipeStaticMeta;
}
@@ -246,6 +278,7 @@ public class PipeStaticMeta {
pipeStaticMeta.sinkParameters.getAttribute().put(key, value);
}
+ pipeStaticMeta.updateCompletionSupport();
return pipeStaticMeta;
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTaskMeta.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTaskMeta.java
index 1e860383edd..73048336400 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTaskMeta.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/meta/PipeTaskMeta.java
@@ -39,6 +39,7 @@ import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
public class PipeTaskMeta {
@@ -60,6 +61,8 @@ public class PipeTaskMeta {
private final Set<PipeRuntimeException> exceptionMessages =
Collections.newSetFromMap(new ConcurrentHashMap<>());
+ private final AtomicLong exceptionMessageVersion = new AtomicLong(0);
+
public PipeTaskMeta(/* @NotNull */ final ProgressIndex progressIndex, final int leaderNodeId) {
this.progressIndex.set(progressIndex);
this.leaderNodeId.set(leaderNodeId);
@@ -119,6 +122,7 @@ public class PipeTaskMeta {
// Here we still keep the map form to allow compatibility with legacy versions
exceptionMessages.clear();
exceptionMessages.add(exceptionMessage);
+ exceptionMessageVersion.incrementAndGet();
}
public synchronized boolean containsExceptionMessage(
@@ -131,11 +135,20 @@ public class PipeTaskMeta {
}
public synchronized void clearExceptionMessages() {
- exceptionMessages.clear();
+ if (!exceptionMessages.isEmpty()) {
+ exceptionMessages.clear();
+ exceptionMessageVersion.incrementAndGet();
+ }
}
public synchronized void clearExceptionMessagesBefore(final long exceptionsClearTime) {
- exceptionMessages.removeIf(exception -> exception.getTimeStamp() <= exceptionsClearTime);
+ if (exceptionMessages.removeIf(exception -> exception.getTimeStamp() <= exceptionsClearTime)) {
+ exceptionMessageVersion.incrementAndGet();
+ }
+ }
+
+ public long getExceptionMessageVersion() {
+ return exceptionMessageVersion.get();
}
public synchronized void serialize(final OutputStream outputStream) throws IOException {
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/PipeEventCommitManager.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/PipeEventCommitManager.java
index 26e7ea305d5..b2654d1e8a7 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/PipeEventCommitManager.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/progress/PipeEventCommitManager.java
@@ -173,6 +173,21 @@ public class PipeEventCommitManager {
return generateCommitterKey(pipeName, creationTime, regionId);
}
+ /**
+ * Checks that the committer is still registered and its restart generation is current. The
+ * registration check rejects dropped tasks; comparing the generated key rejects a barrier from an
+ * older committer after the same Pipe/DataRegion task has been restarted.
+ */
+ public boolean isCurrentCommitterKey(final CommitterKey committerKey) {
+ return committerKey != null
+ && eventCommitterMap.containsKey(committerKey)
+ && committerKey.equals(
+ generateCommitterKey(
+ committerKey.getPipeName(),
+ committerKey.getCreationTime(),
+ committerKey.getRegionId()));
+ }
+
private CommitterKey generateCommitterKey(
final String pipeName, final long creationTime, final int regionId) {
return taskAgent.getCommitterKey(
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java
index 63eedf80a43..6eee75ceb00 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java
@@ -59,6 +59,9 @@ public abstract class EnrichedEvent implements Event {
protected final PipeTaskMeta pipeTaskMeta;
+ // Identifies the source instance that produced this event, including across task restarts.
+ protected long sourceId = Long.MIN_VALUE;
+
protected CommitterKey committerKey;
public static final long NO_COMMIT_ID = -1;
protected long commitId = NO_COMMIT_ID;
@@ -317,6 +320,17 @@ public abstract class EnrichedEvent implements Event {
return creationTime;
}
+ @Override
+ public void bindSource(final long sourceId) {
+ if (this.sourceId == Long.MIN_VALUE) {
+ this.sourceId = sourceId;
+ }
+ }
+
+ public final long getSourceId() {
+ return sourceId;
+ }
+
public final int getRegionId() {
// TODO: persist regionId in EnrichedEvent
return committerKey == null ? -1 : committerKey.getRegionId();
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/ProgressReportEvent.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/ProgressReportEvent.java
index e6e69acac13..3e35522e344 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/ProgressReportEvent.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/ProgressReportEvent.java
@@ -33,6 +33,11 @@ public class ProgressReportEvent extends EnrichedEvent {
private ProgressIndex progressIndex;
+ @Override
+ public boolean isControlEvent() {
+ return true;
+ }
+
public ProgressReportEvent(
final String pipeName, final long creationTime, final PipeTaskMeta pipeTaskMeta) {
super(
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
index b701df3848d..fc41cb852fc 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
@@ -200,6 +200,7 @@ public enum Metric {
PIPE_CONNECTOR_SCHEMA_TRANSFER("pipe_connector_schema_transfer"),
PIPE_DATANODE_REMAINING_EVENT_COUNT("pipe_datanode_remaining_event_count"),
PIPE_DATANODE_REMAINING_TIME("pipe_datanode_remaining_time"),
+ PIPE_DATANODE_COMPLETION_READY("pipe_datanode_completion_ready"),
PIPE_INSERT_NODE_EVENT_TRANSFER_TIME("pipe_insert_node_event_transfer_time"),
PIPE_TSFILE_EVENT_TRANSFER_TIME("pipe_tsfile_event_transfer_time"),
PIPE_DATANODE_EVENT_TRANSFER("pipe_datanode_event_transfer"),
diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/PipeMetaDeSerTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/PipeMetaDeSerTest.java
index eee842621bb..15923f0205f 100644
--- a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/PipeMetaDeSerTest.java
+++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/pipe/task/PipeMetaDeSerTest.java
@@ -29,11 +29,15 @@ import org.apache.iotdb.commons.consensus.index.impl.TimePartitionProgressIndex;
import org.apache.iotdb.commons.consensus.index.impl.TimeWindowStateProgressIndex;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkCriticalException;
+import org.apache.iotdb.commons.pipe.agent.plugin.builtin.BuiltinPipePlugin;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta;
+import org.apache.iotdb.commons.pipe.config.constant.PipeProcessorConstant;
+import org.apache.iotdb.commons.pipe.config.constant.PipeSinkConstant;
+import org.apache.iotdb.commons.pipe.config.constant.PipeSourceConstant;
import org.apache.tsfile.utils.Pair;
import org.apache.tsfile.utils.PublicBAOS;
@@ -41,15 +45,85 @@ import org.apache.tsfile.utils.ReadWriteIOUtils;
import org.junit.Assert;
import org.junit.Test;
+import java.io.ByteArrayInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
+import java.util.Locale;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
public class PipeMetaDeSerTest {
+ @Test
+ public void testCompletionSupportSurvivesBothDeserializationPaths() throws IOException {
+ for (final BuiltinPipePlugin sink :
+ new BuiltinPipePlugin[] {
+ BuiltinPipePlugin.IOTDB_THRIFT_CONNECTOR,
+ BuiltinPipePlugin.IOTDB_THRIFT_ASYNC_SINK,
+ BuiltinPipePlugin.IOTDB_AIR_GAP_CONNECTOR,
+ BuiltinPipePlugin.IOTDB_AIR_GAP_SINK
+ }) {
+ final PipeStaticMeta staticMeta =
+ new PipeStaticMeta(
+ "pipe",
+ 1L,
+ Map.of(),
+ Map.of(),
+ Map.of(PipeSinkConstant.SINK_KEY, sink.getPipePluginName().toUpperCase(Locale.ROOT)));
+ Assert.assertTrue(staticMeta.supportsCompletionMark());
+ Assert.assertTrue(
+ PipeStaticMeta.deserialize(staticMeta.serialize()).supportsCompletionMark());
+ final ByteBuffer serialized = staticMeta.serialize();
+ final byte[] bytes = new byte[serialized.remaining()];
+ serialized.get(bytes);
+ Assert.assertTrue(
+ PipeStaticMeta.deserialize(new ByteArrayInputStream(bytes)).supportsCompletionMark());
+ }
+ }
+
+ @Test
+ public void testCompletionSupportRejectsUnsupportedConfigurations() {
+ Assert.assertFalse(
+ new PipeStaticMeta("__subscription.pipe", 1L, Map.of(), Map.of(), Map.of())
+ .supportsCompletionMark());
+ Assert.assertFalse(
+ new PipeStaticMeta(
+ "pipe",
+ 1L,
+ Map.of(PipeSourceConstant.SOURCE_KEY, "custom-source"),
+ Map.of(),
+ Map.of())
+ .supportsCompletionMark());
+ Assert.assertFalse(
+ new PipeStaticMeta(
+ "pipe",
+ 1L,
+ Map.of(),
+ Map.of(PipeProcessorConstant.PROCESSOR_KEY, "aggregate-processor"),
+ Map.of())
+ .supportsCompletionMark());
+ Assert.assertFalse(
+ new PipeStaticMeta(
+ "pipe",
+ 1L,
+ Map.of(),
+ Map.of(),
+ Map.of(PipeSinkConstant.SINK_KEY, "write-back-sink"))
+ .supportsCompletionMark());
+ Assert.assertFalse(
+ new PipeStaticMeta(
+ "pipe",
+ 1L,
+ Map.of(),
+ Map.of(),
+ Map.of(
+ PipeSinkConstant.SINK_KEY, "iotdb-air-gap-sink",
+ PipeSinkConstant.SINK_LOAD_TSFILE_STRATEGY_KEY, "async"))
+ .supportsCompletionMark());
+ }
+
@Test
public void testPipeStatusTypeCompatibility() {
Assert.assertEquals((byte) 0, PipeStatus.RUNNING.getType());