[Pipe] Avoid no-op consensus writes for covered progress (#18574) (#18684)

(cherry picked from commit 5aae669)

Author: Caideyipi

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParser.java

@@ -211,18 +211,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,
@@ -236,8 +230,8 @@ private void parseHeartbeatAndSaveMetaChangeLocally(
                               + "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);

iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/pipe/coordinator/runtime/heartbeat/PipeHeartbeatParserTest.java

@@ -21,6 +21,8 @@
 
 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.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 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);
@@ -428,6 +507,30 @@ private PipeHeartbeat createPipeHeartbeat(
         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;