IoTV2: Remove events that do not retry in leader's transfer buffer & fix concurrency bug of retryQueue & fix retry interval & refine log (apache#15684)

* remove events that do not retry

* refine log & fix retry interval & fix concurrency issue of retryQueue

Author: Pengzna

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

@@ -72,10 +72,10 @@
 import java.util.Comparator;
 import java.util.Iterator;
 import java.util.Objects;
-import java.util.PriorityQueue;
 import java.util.Queue;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.PriorityBlockingQueue;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -101,7 +101,7 @@ public class PipeConsensusAsyncConnector extends IoTDBConnector implements Conse
   private static final long PIPE_CONSENSUS_EVENT_ENQUEUE_TIMEOUT_IN_MS =
       IOTDB_CONFIG.getConnectionTimeoutInMS() / 6;
   private final Queue<EnrichedEvent> retryEventQueue =
-      new PriorityQueue<>(
+      new PriorityBlockingQueue<>(
           IOTDB_CONFIG.getIotConsensusV2PipelineSize(),
           Comparator.comparingLong(EnrichedEvent::getReplicateIndexForIoTV2));
   // We use enrichedEvent here to make use of EnrichedEvent.equalsInPipeConsensus
@@ -535,9 +535,10 @@ private void asyncTransferQueuedEventsIfNecessary() {
                 ? peekedEvent.getRetryInterval()
                 : 0L;
         LOGGER.info(
-            "PipeConsensus-ConsensusGroup-{}: retry with interval {} for {}",
+            "PipeConsensus-ConsensusGroup-{}: retry with interval {} for index {} {}",
             consensusGroupId,
             retryInterval,
+            peekedEvent.getReplicateIndexForIoTV2(),
             peekedEvent);
         // need to retry in background service, otherwise the retryInterval will block the sender
         // procedure.
@@ -616,7 +617,7 @@ private void retryTransfer(final PipeDeleteDataNodeEvent deleteDataNodeEvent) {
    * @param event event to retry
    */
   @SuppressWarnings("java:S899")
-  public void addFailureEventToRetryQueue(final EnrichedEvent event) {
+  public synchronized void addFailureEventToRetryQueue(final EnrichedEvent event) {
     if (event.isReleased()) {
       return;
     }
@@ -630,11 +631,20 @@ public void addFailureEventToRetryQueue(final EnrichedEvent event) {
       return;
     }
 
-    retryEventQueue.offer(event);
-    LOGGER.info(
-        "PipeConsensus-ConsensusGroup-{}: Event {} transfer failed, will be added to retry queue.",
-        consensusGroupId,
-        event);
+    boolean res = retryEventQueue.offer(event);
+    if (res) {
+      LOGGER.info(
+          "PipeConsensus-ConsensusGroup-{}: Event {} replicate index {} transfer failed, will be added to retry queue.",
+          consensusGroupId,
+          event,
+          event.getReplicateIndexForIoTV2());
+    } else {
+      LOGGER.warn(
+          "PipeConsensus-ConsensusGroup-{}: Event {} replicate index {} transfer failed, added to retry queue failed, this event will be ignored.",
+          consensusGroupId,
+          event,
+          event.getReplicateIndexForIoTV2());
+    }
 
     if (isClosed.get()) {
       event.clearReferenceCount(PipeConsensusAsyncConnector.class.getName());

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

@@ -87,8 +87,10 @@ public void onComplete(TPipeConsensusTransferResp response) {
         LOGGER.info(
             "DeleteNodeTransfer: no.{} event successfully processed!",
             event.getReplicateIndexForIoTV2());
-        connector.removeEventFromBuffer(event);
       }
+      // if code flow reach here, meaning the file will not be resent and will be ignored.
+      // events that don't need to be retried will be removed from the buffer
+      connector.removeEventFromBuffer(event);
 
       long duration = System.nanoTime() - createTime;
       metric.recordConnectorWalTransferTimer(duration);

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

@@ -95,9 +95,12 @@ public void onComplete(TPipeConsensusTransferResp response) {
         LOGGER.info(
             "InsertNodeTransfer: no.{} event successfully processed!",
             ((EnrichedEvent) event).getReplicateIndexForIoTV2());
-        connector.removeEventFromBuffer((EnrichedEvent) event);
       }
 
+      // if code flow reach here, meaning the file will not be resent and will be ignored.
+      // events that don't need to be retried will be removed from the buffer
+      connector.removeEventFromBuffer((EnrichedEvent) event);
+
       long duration = System.nanoTime() - createTime;
       metric.recordConnectorWalTransferTimer(duration);
     } catch (Exception e) {

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

@@ -210,9 +210,9 @@ public void onComplete(final TPipeConsensusTransferResp response) {
                   tsFile.getName());
         }
 
-        if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
-          connector.removeEventFromBuffer(event);
-        }
+        // if code flow reach here, meaning the file will not be resent and will be ignored.
+        // events that don't need to be retried will be removed from the buffer
+        connector.removeEventFromBuffer(event);
       } catch (final Exception e) {
         onError(e);
         return;
@@ -295,10 +295,10 @@ public void onError(final Exception exception) {
 
     if (RetryUtils.needRetryWithIncreasingInterval(exception)) {
       // just in case for overflow
-      if (event.getRetryInterval() << 2 <= 0) {
+      if (event.getRetryInterval() << 1 <= 0) {
         event.setRetryInterval(1000L * 20);
       } else {
-        event.setRetryInterval(Math.min(1000L * 20, event.getRetryInterval() << 2));
+        event.setRetryInterval(Math.min(1000L * 20, event.getRetryInterval() << 1));
       }
     }
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/event/realtime/PipeRealtimeEventFactory.java

@@ -61,7 +61,7 @@ public static PipeRealtimeEvent createRealtimeEvent(
         && PipeConsensusProcessor.isShouldReplicate(tsFileInsertionEvent)) {
       tsFileInsertionEvent.setReplicateIndexForIoTV2(
           ReplicateProgressDataNodeManager.assignReplicateIndexForIoTV2(dataRegionId));
-      LOGGER.debug(
+      LOGGER.info(
           "[Region{}]Set {} for event {}",
           dataRegionId,
           tsFileInsertionEvent.getReplicateIndexForIoTV2(),
@@ -103,7 +103,7 @@ public static PipeRealtimeEvent createRealtimeEvent(
         && PipeConsensusProcessor.isShouldReplicate(insertionEvent)) {
       insertionEvent.setReplicateIndexForIoTV2(
           ReplicateProgressDataNodeManager.assignReplicateIndexForIoTV2(dataRegionId));
-      LOGGER.debug(
+      LOGGER.info(
           "[Region{}]Set {} for event {}",
           dataRegionId,
           insertionEvent.getReplicateIndexForIoTV2(),
@@ -130,7 +130,7 @@ public static PipeRealtimeEvent createRealtimeEvent(
         && PipeConsensusProcessor.isShouldReplicate(deleteDataNodeEvent)) {
       deleteDataNodeEvent.setReplicateIndexForIoTV2(
           ReplicateProgressDataNodeManager.assignReplicateIndexForIoTV2(dataRegionId));
-      LOGGER.debug(
+      LOGGER.info(
           "[Region{}]Set {} for event {}",
           dataRegionId,
           deleteDataNodeEvent.getReplicateIndexForIoTV2(),

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/extractor/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionExtractor.java

@@ -879,7 +879,7 @@ private Event supplyTsFileEvent(final TsFileResource resource) {
       event.setReplicateIndexForIoTV2(
           ReplicateProgressDataNodeManager.assignReplicateIndexForIoTV2(
               resource.getDataRegionId()));
-      LOGGER.debug(
+      LOGGER.info(
           "[Region{}]Set {} for event {}",
           resource.getDataRegionId(),
           event.getReplicateIndexForIoTV2(),

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/event/EnrichedEvent.java

@@ -435,8 +435,8 @@ public long getRetryInterval() {
     return this.retryInterval;
   }
 
-  public long setRetryInterval(final long retryInterval) {
-    return retryInterval;
+  public void setRetryInterval(final long retryInterval) {
+    this.retryInterval = retryInterval;
   }
 
   public CommitterKey getCommitterKey() {