This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/dev/1.3 by this push:
new a42ea12c871 [Pipe] Avoid no-op consensus writes for covered progress
(#18574) (#18684)
a42ea12c871 is described below
commit a42ea12c871572de634bd6b928bbfdb88dc58988
Author: Caideyipi <[email protected]>
AuthorDate: Tue Sep 22 15:36:13 2026 +0800
[Pipe] Avoid no-op consensus writes for covered progress (#18574) (#18684)
(cherry picked from commit 5aae6697fba555c40a3bba40529ef8e7f8a2e2c2)
---
.../runtime/heartbeat/PipeHeartbeatParser.java | 20 ++--
.../runtime/heartbeat/PipeHeartbeatParserTest.java | 103 +++++++++++++++++++++
2 files changed, 110 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 19b0b73e674..c617a37af1e 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
@@ -211,18 +211,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,
@@ -236,8 +230,8 @@ public class PipeHeartbeatParser {
+ "Progress index on coordinator: {}, progress
index from agent: {}, updated progressIndex: {}",
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 ccc12244297..cd0e03654cb 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
@@ -21,6 +21,8 @@ package
org.apache.iotdb.confignode.manager.pipe.coordinator.runtime.heartbeat;
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;
@@ -55,6 +57,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
+import static org.junit.Assert.assertEquals;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyBoolean;
import static org.mockito.Mockito.never;
@@ -134,6 +137,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);
@@ -428,6 +507,30 @@ public class PipeHeartbeatParserTest {
Collections.singletonList(recentFailures));
}
+ private PipeHeartbeat createPipeHeartbeat(final PipeMeta pipeMeta, final
boolean ignored)
+ throws Exception {
+ return createPipeHeartbeat(pipeMeta, Collections.emptyMap());
+ }
+
+ private PipeMeta createPipeMeta() {
+ final PipeRuntimeMeta pipeRuntimeMeta = new PipeRuntimeMeta();
+ pipeRuntimeMeta
+ .getConsensusGroupId2TaskMetaMap()
+ .put(DATA_NODE_ID, new PipeTaskMeta(MinimumProgressIndex.INSTANCE,
DATA_NODE_ID));
+ return new PipeMeta(
+ new PipeStaticMeta(
+ "test_pipe", 1L, Collections.emptyMap(), new HashMap<>(), new
HashMap<>()),
+ pipeRuntimeMeta);
+ }
+
+ 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 static class ParserTestContext {
private final PipeHeartbeatParser parser;
private final ProcedureManager procedureManager;
