From 13c7d5f6017919cbe00ca9c719e87bfe89eadbb4 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 3 Sep 2026 11:26:19 +0800 Subject: [PATCH] [Pipe] Avoid no-op consensus writes for covered progress --- .../heartbeat/PipeHeartbeatParser.java | 20 ++--- .../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 f83749348115..a6b1e95ae3cc 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 @@ private void parseHeartbeatAndSaveMetaChangeLocally( } // 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 @@ private void parseHeartbeatAndSaveMetaChangeLocally( .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 b76156bb6192..ef4c03549afc 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.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 void testParseHeartbeatKeepsPendingFlagsWhenProcedureSubmissionFails() th 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 @@ private PipeMeta createHistoryOnlyPipeMeta(final int... regionIds) { return createPipeMeta(sourceAttributes, regionIds); } + private RecoverProgressIndex createRecoverProgressIndex( + final long firstDataNodeIndex, final long secondDataNodeIndex) { + final Map 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 sourceAttributes, final int... regionIds) { final PipeRuntimeMeta pipeRuntimeMeta = new PipeRuntimeMeta();