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 5aae6697fba [Pipe] Avoid no-op consensus writes for covered progress
(#18574)
5aae6697fba is described below
commit 5aae6697fba555c40a3bba40529ef8e7f8a2e2c2
Author: Caideyipi <[email protected]>
AuthorDate: Mon Sep 21 10:45:36 2026 +0800
[Pipe] Avoid no-op consensus writes for covered progress (#18574)
---
.../runtime/heartbeat/PipeHeartbeatParser.java | 20 ++---
.../runtime/heartbeat/PipeHeartbeatParserTest.java | 86 ++++++++++++++++++++++
2 files changed, 93 insertions(+), 13 deletions(-)
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java
index f8374934811..a6b1e95ae3c 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java
@@ -236,18 +236,12 @@ public class PipeHeartbeatParser {
}
// Update progress index
- if (!(runtimeMetaFromCoordinator
- .getValue()
- .getProgressIndex()
- .isAfter(runtimeMetaFromAgent.getProgressIndex())
- || runtimeMetaFromCoordinator
- .getValue()
- .getProgressIndex()
- .equals(runtimeMetaFromAgent.getProgressIndex()))) {
+ final ProgressIndex coordinatorProgressIndex =
+ runtimeMetaFromCoordinator.getValue().getProgressIndex();
+ final ProgressIndex agentProgressIndex =
runtimeMetaFromAgent.getProgressIndex();
+ if (!coordinatorProgressIndex.isEqualOrAfter(agentProgressIndex)) {
final ProgressIndex updatedProgressIndex =
- runtimeMetaFromCoordinator
- .getValue()
-
.updateProgressIndex(runtimeMetaFromAgent.getProgressIndex());
+
runtimeMetaFromCoordinator.getValue().updateProgressIndex(agentProgressIndex);
PipeConfigNodeResourceManager.log()
.schedule(
PipeHeartbeatParser.class,
@@ -263,8 +257,8 @@ public class PipeHeartbeatParser {
.LOG_PROGRESS_INDEX_COORDINATOR_ARG_PROGRESS_INDEX_AGENT_ARG_UPDATED_PROGRESSINDEX_1A22ABC5,
pipeMetaFromCoordinator.getStaticMeta().getPipeName(),
runtimeMetaFromCoordinator.getKey(),
-
runtimeMetaFromCoordinator.getValue().getProgressIndex(),
- runtimeMetaFromAgent.getProgressIndex(),
+ coordinatorProgressIndex,
+ agentProgressIndex,
updatedProgressIndex));
needWriteConsensusOnConfigNodes.set(true);
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParserTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParserTest.java
index b76156bb619..ef4c03549af 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParserTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParserTest.java
@@ -23,6 +23,8 @@ import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId;
import org.apache.iotdb.common.rpc.thrift.TPipeCompletedDataRegion;
import org.apache.iotdb.commons.conf.CommonDescriptor;
import org.apache.iotdb.commons.consensus.index.impl.MinimumProgressIndex;
+import org.apache.iotdb.commons.consensus.index.impl.RecoverProgressIndex;
+import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex;
import org.apache.iotdb.commons.exception.pipe.PipeRuntimeCriticalException;
import
org.apache.iotdb.commons.exception.pipe.PipeRuntimeSinkCriticalException;
import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta;
@@ -141,6 +143,82 @@ public class PipeHeartbeatParserTest {
verify(context.procedureManager, times(2)).pipeHandleMetaChange(true,
false);
}
+ @Test
+ public void
testParseHeartbeatSkipsConsensusWriteWhenCoordinatorProgressCoversAgent()
+ throws Exception {
+
CommonDescriptor.getInstance().getConfig().setSeperatedPipeHeartbeatEnabled(false);
+
+ final PipeTaskInfo pipeTaskInfo = new PipeTaskInfo();
+ final PipeMeta coordinatorPipeMeta = createPipeMeta();
+ final RecoverProgressIndex coordinatorProgressIndex =
createRecoverProgressIndex(10, 10);
+ coordinatorPipeMeta
+ .getRuntimeMeta()
+ .getConsensusGroupId2TaskMetaMap()
+ .get(DATA_NODE_ID)
+ .updateProgressIndex(coordinatorProgressIndex);
+ pipeTaskInfo.createPipe(
+ new CreatePipePlanV2(
+ coordinatorPipeMeta.getStaticMeta(),
coordinatorPipeMeta.getRuntimeMeta()));
+
+ final PipeMeta agentPipeMeta = createPipeMeta();
+ agentPipeMeta
+ .getRuntimeMeta()
+ .getConsensusGroupId2TaskMetaMap()
+ .get(DATA_NODE_ID)
+ .updateProgressIndex(createRecoverProgressIndex(10, 5));
+
+ final ParserTestContext context = createParserTestContext(1, pipeTaskInfo);
+ context.parser.parseHeartbeat(DATA_NODE_ID,
createPipeHeartbeat(agentPipeMeta, false));
+
+ assertEquals(
+ coordinatorProgressIndex,
+ pipeTaskInfo
+ .getPipeMetaByPipeName("test_pipe")
+ .getRuntimeMeta()
+ .getConsensusGroupId2TaskMetaMap()
+ .get(DATA_NODE_ID)
+ .getProgressIndex());
+ verify(context.procedureManager,
never()).pipeHandleMetaChange(anyBoolean(), anyBoolean());
+ }
+
+ @Test
+ public void
testParseHeartbeatWritesConsensusWhenAgentProgressAdvancesCoordinator()
+ throws Exception {
+
CommonDescriptor.getInstance().getConfig().setSeperatedPipeHeartbeatEnabled(false);
+
+ final PipeTaskInfo pipeTaskInfo = new PipeTaskInfo();
+ final PipeMeta coordinatorPipeMeta = createPipeMeta();
+ coordinatorPipeMeta
+ .getRuntimeMeta()
+ .getConsensusGroupId2TaskMetaMap()
+ .get(DATA_NODE_ID)
+ .updateProgressIndex(createRecoverProgressIndex(10, 10));
+ pipeTaskInfo.createPipe(
+ new CreatePipePlanV2(
+ coordinatorPipeMeta.getStaticMeta(),
coordinatorPipeMeta.getRuntimeMeta()));
+
+ final PipeMeta agentPipeMeta = createPipeMeta();
+ final RecoverProgressIndex agentProgressIndex =
createRecoverProgressIndex(10, 11);
+ agentPipeMeta
+ .getRuntimeMeta()
+ .getConsensusGroupId2TaskMetaMap()
+ .get(DATA_NODE_ID)
+ .updateProgressIndex(agentProgressIndex);
+
+ final ParserTestContext context = createParserTestContext(1, pipeTaskInfo);
+ context.parser.parseHeartbeat(DATA_NODE_ID,
createPipeHeartbeat(agentPipeMeta, false));
+
+ assertEquals(
+ agentProgressIndex,
+ pipeTaskInfo
+ .getPipeMetaByPipeName("test_pipe")
+ .getRuntimeMeta()
+ .getConsensusGroupId2TaskMetaMap()
+ .get(DATA_NODE_ID)
+ .getProgressIndex());
+ verify(context.procedureManager, times(1)).pipeHandleMetaChange(true,
false);
+ }
+
@Test
public void testParseHeartbeatIgnoresExceptionsBeforeClearTime() throws
Exception {
CommonDescriptor.getInstance().getConfig().setSeperatedPipeHeartbeatEnabled(false);
@@ -683,6 +761,14 @@ public class PipeHeartbeatParserTest {
return createPipeMeta(sourceAttributes, regionIds);
}
+ private RecoverProgressIndex createRecoverProgressIndex(
+ final long firstDataNodeIndex, final long secondDataNodeIndex) {
+ final Map<Integer, SimpleProgressIndex> dataNodeId2LocalIndex = new
HashMap<>();
+ dataNodeId2LocalIndex.put(1, new SimpleProgressIndex(0,
firstDataNodeIndex));
+ dataNodeId2LocalIndex.put(2, new SimpleProgressIndex(0,
secondDataNodeIndex));
+ return new RecoverProgressIndex(dataNodeId2LocalIndex);
+ }
+
private PipeMeta createPipeMeta(
final Map<String, String> sourceAttributes, final int... regionIds) {
final PipeRuntimeMeta pipeRuntimeMeta = new PipeRuntimeMeta();
