[IoTV2]: Pick deletion event for historical resend

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2AsyncSink.java

@@ -236,7 +236,15 @@ public synchronized void removeEventFromBuffer(EnrichedEvent event) {
     while (!current.equalsInIoTConsensusV2(event) && iterator.hasNext()) {
       current = iterator.next();
     }
-    iterator.remove();
+    if (current.equalsInIoTConsensusV2(event)) {
+      iterator.remove();
+    } else {
+      LOGGER.warn(
+          "IoTConsensusV2-ConsensusGroup-{}: event-{} not found in transferBuffer, skip removing. queue size = {}",
+          consensusGroupId,
+          event,
+          transferBuffer.size());
+    }
     // update replicate progress
     currentReplicateProgress =
         Math.max(currentReplicateProgress, event.getReplicateIndexForIoTV2());

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java

@@ -930,6 +930,18 @@ private Event supplyDeletionEvent(final DeletionResource deletionResource) {
             skipIfNoPrivileges,
             false);
 
+    // if using IoTV2, assign a replicateIndex for this historical deletion event
+    if (DataRegionConsensusImpl.getInstance() instanceof IoTConsensusV2
+        && IoTConsensusV2Processor.isShouldReplicate(event)) {
+      event.setReplicateIndexForIoTV2(
+          ReplicateProgressDataNodeManager.assignReplicateIndexForIoTV2(pipeName));
+      LOGGER.debug(
+          "[{}]Set {} for historical deletion event {}",
+          pipeName,
+          event.getReplicateIndexForIoTV2(),
+          event);
+    }
+
     if (sloppyPattern || isDbNameCoveredByPattern) {
       event.skipParsingPattern();
     }