This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new a7e508cf458 [Subscription] Index consensus ACK progress by writer
(#18835)
a7e508cf458 is described below
commit a7e508cf458857cc4afa1e661220d099559b67f9
Author: Caideyipi <[email protected]>
AuthorDate: Wed Oct 14 11:20:48 2026 +0800
[Subscription] Index consensus ACK progress by writer (#18835)
---
.../ConsensusSubscriptionCommitManager.java | 93 ++++++++++++++--------
.../ConsensusSubscriptionCommitStateTest.java | 20 +++++
2 files changed, 79 insertions(+), 34 deletions(-)
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 7d5edd25828..14e50ae44d0 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.StandardCopyOption;
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 @@ public class ConsensusSubscriptionCommitManager {
/** Tracks dispatched but not-yet-committed events by writer-local slot. */
private final Map<ProgressSlot, ProgressKey> outstandingKeys = new
ConcurrentHashMap<>();
- /** Tracks committed dispatched entries that cannot yet advance the
frontier because of gaps. */
- private final Map<ProgressSlot, ProgressKey> 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<Integer, NavigableSet<ProgressKey>>
outstandingKeysByWriter = new HashMap<>();
+
+ /** Tracks ACKed entries blocked by earlier outstanding progress from the
same writer. */
+ private final Map<Integer, NavigableSet<ProgressKey>>
committedPendingKeysByWriter =
+ new HashMap<>();
public ConsensusSubscriptionCommitState(
final String regionId, final SubscriptionConsensusProgress progress) {
@@ -1675,6 +1684,10 @@ public class ConsensusSubscriptionCommitManager {
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 @@ public class ConsensusSubscriptionCommitManager {
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 @@ public class ConsensusSubscriptionCommitManager {
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 @@ public class ConsensusSubscriptionCommitManager {
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 @@ public class ConsensusSubscriptionCommitManager {
}
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 @@ public class ConsensusSubscriptionCommitManager {
if (Objects.nonNull(nextOutstanding) &&
nextOutstanding.compareTo(nextCommitted) < 0) {
return;
}
- committedPendingKeys.remove(ProgressSlot.from(nextCommitted));
+ removeCommittedPendingKey(nextCommitted);
advanceCommitted(nextCommitted);
current = nextCommitted;
}
@@ -1897,39 +1911,50 @@ public class ConsensusSubscriptionCommitManager {
}
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<Integer, NavigableSet<ProgressKey>> index, final ProgressKey
key) {
+ index.computeIfAbsent(key.writerNodeId, ignored -> new
TreeSet<>()).add(key);
+ }
+
+ private static void removeFromWriterIndex(
+ final Map<Integer, NavigableSet<ProgressKey>> index, final ProgressKey
key) {
+ final NavigableSet<ProgressKey> 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<Integer, NavigableSet<ProgressKey>> index,
+ final WriterId writerId,
+ final ProgressKey current) {
+ if (Objects.isNull(writerId)) {
+ return null;
+ }
+ final NavigableSet<ProgressKey> 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 75aa3032307..24778e2c175 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 class ConsensusSubscriptionCommitStateTest {
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 =
