[IoTV2]: Pick deletion event for historical resend

[Pengzna]

1. Historical deletion events missing replicateIndex, silently dropped by Processor.
2. RemoveEventFromBuffer removes wrong event when target not found.

[Copilot]

Pull request overview

This PR addresses IoTConsensusV2 historical resend correctness by ensuring historical deletion events participate in IoTV2 replication tracking, and by preventing accidental removal of the wrong buffered event in the async sink.

Changes:

• Assign replicateIndexForIoTV2 to historical deletion events so they aren’t dropped by IoTConsensusV2Processor.
• Fix IoTConsensusV2AsyncSink.removeEventFromBuffer to avoid removing an unrelated event when the target event is not found.

Reviewed changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated no comments.

File	Description
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/source/dataregion/historical/PipeHistoricalDataRegionTsFileAndDeletionSource.java	Assigns IoTV2 replicateIndex for historical deletion events to ensure they can be forwarded by the IoTV2 processor.
iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/iotconsensusv2/IoTConsensusV2AsyncSink.java	Prevents removing the wrong element from transferBuffer when the requested event is not present.

Comments suppressed due to low confidence (1)

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

• removeEventFromBuffer now correctly avoids removing the wrong element when the target event isn’t found, but it still updates currentReplicateProgress and (more importantly) calls event.decreaseReferenceCount(...) even in the not-found branch. If the event is still in transferBuffer (e.g., equals mismatch) this can prematurely release resources while the queued event remains; if it was already removed earlier it can also drive the reference count negative. Consider only decreasing the connector-held reference (and possibly updating progress) when the event was actually removed, or alternatively remove/decrease based on the exact buffered element found.

    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());
    // decrease reference count
    event.decreaseReferenceCount(IoTConsensusV2AsyncSink.class.getName(), true);

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.