PipeConsensus: Avoid replicate block && Avoid transfer error when con…

…nector is closed. (#13146)

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/connector/protocol/pipeconsensus/PipeConsensusAsyncConnector.java

@@ -167,6 +167,11 @@ private boolean addEvent2Buffer(EnrichedEvent event) {
             event.getCommitId(),
             event);
       }
+      // Special judge to avoid transfer stuck when re-transfer events that will not be put in
+      // retryQueue.
+      if (transferBuffer.contains(event)) {
+        return true;
+      }
       long currentTime = System.nanoTime();
       boolean result =
           transferBuffer.offer(
@@ -203,6 +208,13 @@ public synchronized void removeEventFromBuffer(EnrichedEvent event) {
           transferBuffer.size(),
           IOTDB_CONFIG.getPipeConsensusPipelineSize());
     }
+    if (transferBuffer.isEmpty()) {
+      LOGGER.info(
+          "PipeConsensus-ConsensusGroup-{}: try to remove event-{} after pipeConsensusAsyncConnector being closed. Ignore it.",
+          consensusGroupId,
+          event);
+      return;
+    }
     Iterator<EnrichedEvent> iterator = transferBuffer.iterator();
     EnrichedEvent current = iterator.next();
     while (!current.equalsInPipeConsensus(event) && iterator.hasNext()) {
@@ -458,7 +470,7 @@ private void syncTransferQueuedEventsIfNecessary() throws Exception {
                 polledEvent);
           }
         }
-        if (polledEvent != null && LOGGER.isDebugEnabled()) {
+        if (polledEvent != null) {
           if (LOGGER.isDebugEnabled()) {
             LOGGER.debug("Polled event {} from retry queue.", polledEvent);
           }