diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java index 7d5edd25828e..14e50ae44d0a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitManager.java @@ -67,13 +67,16 @@ import java.nio.file.StandardOpenOption; import java.util.Base64; import java.util.Collections; +import java.util.HashMap; import java.util.Iterator; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; +import java.util.NavigableSet; import java.util.Objects; import java.util.Set; import java.util.TreeMap; +import java.util.TreeSet; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; @@ -1617,8 +1620,14 @@ protected boolean removeEldestEntry(final Map.Entry eldest /** Tracks dispatched but not-yet-committed events by writer-local slot. */ private final Map outstandingKeys = new ConcurrentHashMap<>(); - /** Tracks committed dispatched entries that cannot yet advance the frontier because of gaps. */ - private final Map committedPendingKeys = new LinkedHashMap<>(); + // Exact ACK lookup uses the slot map above. Successor lookup must not scan the entire backlog + // on every ACK or every step of a contiguous advance. Both indexes use ProgressKey's existing + // physical-time/node/sequence ordering and are guarded by the same state monitor as the maps. + private final Map> outstandingKeysByWriter = new HashMap<>(); + + /** Tracks ACKed entries blocked by earlier outstanding progress from the same writer. */ + private final Map> committedPendingKeysByWriter = + new HashMap<>(); public ConsensusSubscriptionCommitState( final String regionId, final SubscriptionConsensusProgress progress) { @@ -1675,6 +1684,10 @@ public void recordMapping(final WriterId writerId, final WriterProgress writerPr final ProgressSlot slot = ProgressSlot.from(key); synchronized (this) { final ProgressKey previous = outstandingKeys.put(slot, key); + if (Objects.nonNull(previous)) { + removeFromWriterIndex(outstandingKeysByWriter, previous); + } + addToWriterIndex(outstandingKeysByWriter, key); if (Objects.nonNull(previous) && !previous.equals(key)) { LOGGER.warn( DataNodePipeMessages @@ -1720,7 +1733,7 @@ CommitOperationResult commitAndGetResult( final ProgressKey key = new ProgressKey(writerId, writerProgress); synchronized (this) { - final ProgressKey recordedKey = outstandingKeys.remove(ProgressSlot.from(key)); + final ProgressKey recordedKey = removeOutstandingKey(key); if (recordedKey == null) { if (recentlyCommittedKeys.contains(key)) { LOGGER.debug( @@ -1781,7 +1794,7 @@ CommitOperationResult commitWithoutOutstandingAndGetResult( return CommitOperationResult.handledWithoutAdvance(); } - final ProgressKey outstandingKey = outstandingKeys.remove(ProgressSlot.from(incomingKey)); + final ProgressKey outstandingKey = removeOutstandingKey(incomingKey); if (Objects.isNull(outstandingKey)) { // Late or duplicate ACKs are reported by the queue or summarized by the receiver. Logging // every missing mapping at WARN would amplify one commit request by its context count. @@ -1809,7 +1822,8 @@ CommitOperationResult commitWithoutOutstandingAndGetResult( public void resetForSeek(final RegionProgress regionProgress) { synchronized (this) { outstandingKeys.clear(); - committedPendingKeys.clear(); + outstandingKeysByWriter.clear(); + committedPendingKeysByWriter.clear(); recentlyCommittedKeys.clear(); committedWriterPositions.clear(); if (Objects.nonNull(regionProgress)) { @@ -1861,10 +1875,10 @@ private WriterProgress getCommittedWriterProgressForWriter(final WriterId writer } private void stageCommittedAndAdvance(final ProgressKey key) { - committedPendingKeys.put(ProgressSlot.from(key), key); + addToWriterIndex(committedPendingKeysByWriter, key); final WriterId writerId = key.toWriterId(regionId); if (Objects.isNull(writerId)) { - committedPendingKeys.remove(ProgressSlot.from(key)); + removeCommittedPendingKey(key); return; } ProgressKey current = @@ -1878,7 +1892,7 @@ private void stageCommittedAndAdvance(final ProgressKey key) { if (Objects.nonNull(nextOutstanding) && nextOutstanding.compareTo(nextCommitted) < 0) { return; } - committedPendingKeys.remove(ProgressSlot.from(nextCommitted)); + removeCommittedPendingKey(nextCommitted); advanceCommitted(nextCommitted); current = nextCommitted; } @@ -1897,39 +1911,50 @@ private void advanceCommittedIfAhead(final ProgressKey key) { } private ProgressKey findNextCommittedKey(final WriterId writerId, final ProgressKey current) { - ProgressKey next = null; - for (final ProgressKey candidate : committedPendingKeys.values()) { - if (!sameWriter(writerId, candidate)) { - continue; - } - if (candidate.compareTo(current) <= 0) { - continue; - } - if (Objects.isNull(next) || candidate.compareTo(next) < 0) { - next = candidate; - } - } - return next; + return findNextKey(committedPendingKeysByWriter, writerId, current); } private ProgressKey findNextOutstandingKey(final WriterId writerId, final ProgressKey current) { - ProgressKey next = null; - for (final ProgressKey candidate : outstandingKeys.values()) { - if (!sameWriter(writerId, candidate)) { - continue; - } - if (candidate.compareTo(current) <= 0) { - continue; - } - if (Objects.isNull(next) || candidate.compareTo(next) < 0) { - next = candidate; + return findNextKey(outstandingKeysByWriter, writerId, current); + } + + private ProgressKey removeOutstandingKey(final ProgressKey key) { + final ProgressKey removed = outstandingKeys.remove(ProgressSlot.from(key)); + if (Objects.nonNull(removed)) { + removeFromWriterIndex(outstandingKeysByWriter, removed); + } + return removed; + } + + private void removeCommittedPendingKey(final ProgressKey key) { + removeFromWriterIndex(committedPendingKeysByWriter, key); + } + + private static void addToWriterIndex( + final Map> index, final ProgressKey key) { + index.computeIfAbsent(key.writerNodeId, ignored -> new TreeSet<>()).add(key); + } + + private static void removeFromWriterIndex( + final Map> index, final ProgressKey key) { + final NavigableSet writerKeys = index.get(key.writerNodeId); + if (Objects.nonNull(writerKeys)) { + writerKeys.remove(key); + if (writerKeys.isEmpty()) { + index.remove(key.writerNodeId); } } - return next; } - private boolean sameWriter(final WriterId writerId, final ProgressKey key) { - return Objects.nonNull(writerId) && writerId.getNodeId() == key.writerNodeId; + private static ProgressKey findNextKey( + final Map> index, + final WriterId writerId, + final ProgressKey current) { + if (Objects.isNull(writerId)) { + return null; + } + final NavigableSet writerKeys = index.get(writerId.getNodeId()); + return Objects.nonNull(writerKeys) ? writerKeys.higher(current) : null; } private CommitOperationResult buildCommitOperationResult( diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java index 75aa30323078..24778e2c175f 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionCommitStateTest.java @@ -86,6 +86,26 @@ public void testCommitAdvancesContiguousWriterProgress() { assertEquals(7, state.getCommittedWriterNodeId()); } + @Test + public void testReverseAckWithLargeBacklogAdvancesToLatestProgress() { + final WriterId writerId = new WriterId("1_2", 7); + final ConsensusSubscriptionCommitManager.ConsensusSubscriptionCommitState state = + new ConsensusSubscriptionCommitManager.ConsensusSubscriptionCommitState( + "1_2", new SubscriptionConsensusProgress()); + final int backlogSize = 2048; + + for (int localSeq = 0; localSeq < backlogSize; localSeq++) { + state.recordMapping(writerId, new WriterProgress(1_000L + localSeq, localSeq)); + } + for (int localSeq = backlogSize - 1; localSeq >= 0; localSeq--) { + assertTrue(state.commit(writerId, new WriterProgress(1_000L + localSeq, localSeq))); + } + + assertEquals( + new WriterProgress(1_000L + backlogSize - 1, backlogSize - 1L), + state.getCommittedRegionProgress().getWriterPositions().get(writerId)); + } + @Test public void testSerializeDeserializeWriterProgress() throws Exception { final ConsensusSubscriptionCommitManager.ConsensusSubscriptionCommitState state =