From e69549b9165d300a825e0a2fd3d70eb290f67932 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 13 Oct 2026 12:52:55 +0800 Subject: [PATCH 1/2] Optimize subscription WAL acknowledgement path --- .../dataregion/wal/io/WALMetaData.java | 13 +++++-- .../consensus/ConsensusPrefetchingQueue.java | 38 ++++++++++--------- 2 files changed, 29 insertions(+), 22 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java index 50ae0961ce1ff..1407f106dae58 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java @@ -82,12 +82,17 @@ public WALMetaData() { } public WALMetaData(long firstSearchIndex, List buffersSize, Set memTablesId) { + this(firstSearchIndex, buffersSize, memTablesId, 0); + } + + private WALMetaData( + long firstSearchIndex, List buffersSize, Set memTablesId, int entriesNum) { this.firstSearchIndex = firstSearchIndex; this.buffersSize = buffersSize; this.memTablesId = memTablesId; - this.physicalTimes = new ArrayList<>(); - this.nodeIds = new ArrayList<>(); - this.localSeqs = new ArrayList<>(); + this.physicalTimes = new ArrayList<>(entriesNum); + this.nodeIds = new ArrayList<>(entriesNum); + this.localSeqs = new ArrayList<>(entriesNum); } /** Adds an entry without explicit writer progress metadata. */ @@ -248,7 +253,7 @@ public static WALMetaData deserialize(ByteBuffer buffer, WALFileVersion version) memTablesId.add(buffer.getLong()); } } - WALMetaData result = new WALMetaData(firstSearchIndex, buffersSize, memTablesId); + WALMetaData result = new WALMetaData(firstSearchIndex, buffersSize, memTablesId, entriesNum); // V3 extension: file-level timestamp range + per-entry writer progress metadata if (version == WALFileVersion.V3 && buffer.hasRemaining()) { result.minDataTs = buffer.getLong(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java index dd5d204000970..cf12658a6e361 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java @@ -3108,6 +3108,7 @@ private InFlightAckResult ackInFlightEvent( final WriterProgress commitWriterProgress = extractCommitWriterProgress(commitContext); final AtomicBoolean found = new AtomicBoolean(false); final AtomicBoolean acked = new AtomicBoolean(false); + final SubscriptionEvent[] eventToCommit = new SubscriptionEvent[1]; inFlightEvents.compute( new InFlightEventKey(consumerId, commitContext), (key, ev) -> { @@ -3128,25 +3129,26 @@ private InFlightAckResult ackInFlightEvent( return null; } - final boolean committed = - commitAndRefreshWalRetention(commitWriterId, commitWriterProgress); - if (!committed) { - if (!silent) { - LOGGER.warn( - DataNodePipeMessages - .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_FAILED_TO_ADVANCE_COMMIT_FRONTIER_56E606C0, - this, - commitContext); - } - return ev; - } - - ev.ack(); - ev.recordCommittedTimestamp(); - acked.set(true); - cleanUpEvent(ev, false); - return null; + eventToCommit[0] = ev; + return ev; }); + final SubscriptionEvent event = eventToCommit[0]; + if (event != null) { + final boolean committed = commitAndRefreshWalRetention(commitWriterId, commitWriterProgress); + if (committed + && inFlightEvents.remove(new InFlightEventKey(consumerId, commitContext), event)) { + event.ack(); + event.recordCommittedTimestamp(); + acked.set(true); + cleanUpEvent(event, false); + } else if (!committed && !silent) { + LOGGER.warn( + DataNodePipeMessages + .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_FAILED_TO_ADVANCE_COMMIT_FRONTIER_56E606C0, + this, + commitContext); + } + } return !found.get() ? InFlightAckResult.MISSING : acked.get() ? InFlightAckResult.ACKED : InFlightAckResult.REJECTED; From abef0a8c9da3f2e187d82f8bc3a53c91df1d6aa3 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Fri, 23 Oct 2026 16:07:28 +0800 Subject: [PATCH 2/2] Fix subscription ACK cleanup ordering --- .../dataregion/wal/io/WALMetaData.java | 7 ++++- .../consensus/ConsensusPrefetchingQueue.java | 27 +++++++++++-------- .../ConsensusPrefetchingQueueTest.java | 1 + 3 files changed, 23 insertions(+), 12 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java index ea875aefd7c21..32a1c37d08b1f 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/wal/io/WALMetaData.java @@ -254,7 +254,12 @@ public static WALMetaData deserialize(ByteBuffer buffer, WALFileVersion version) memTablesId.add(buffer.getLong()); } } - WALMetaData result = new WALMetaData(firstSearchIndex, buffersSize, memTablesId, entriesNum); + WALMetaData result = + new WALMetaData( + firstSearchIndex, + buffersSize, + memTablesId, + version == WALFileVersion.V3 ? entriesNum : 0); // V3 extension: file-level timestamp range + per-entry writer progress metadata if (version == WALFileVersion.V3 && buffer.hasRemaining()) { result.minDataTs = buffer.getLong(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java index 04e791a079368..a243a3969e438 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueue.java @@ -4147,18 +4147,23 @@ private InFlightAckResult ackInFlightEvent( final SubscriptionEvent event = eventToCommit[0]; if (event != null) { try { - final boolean committed = - commitAndRefreshWalRetention(commitWriterId, commitWriterProgress); + final boolean committed = commitWriterProgress(commitWriterId, commitWriterProgress); if (committed) { - synchronized (event) { - event.ack(); - event.recordCommittedTimestamp(); - inFlightEvents.remove(new InFlightEventKey(consumerId, commitContext), event); - inFlightEvents.entrySet().removeIf(entry -> entry.getValue() == event); - prefetchingQueue.remove(event); - acked.set(true); - cleanUpEvent(event, false); - } + final InFlightEventKey key = new InFlightEventKey(consumerId, commitContext); + inFlightEvents.compute( + key, + (ignored, inFlightEvent) -> { + synchronized (event) { + event.ack(); + event.recordCommittedTimestamp(); + acked.set(true); + cleanUpEvent(event, false); + } + return inFlightEvent == event ? null : inFlightEvent; + }); + inFlightEvents.entrySet().removeIf(entry -> entry.getValue() == event); + prefetchingQueue.remove(event); + refreshCommittedWalRetentionBoundAndNotify(); } else if (!silent) { LOGGER.warn( DataNodePipeMessages diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java index 8d7edd979c50f..c5294d2b6ba5b 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueTest.java @@ -2686,6 +2686,7 @@ ProgressWALReader openProgressWALReader(final File walFile) throws IOException { // A slow retention read must not keep the event-map bucket locked by ACK and stall // a shared prefetch worker trying to recycle that event. executor.submit(() -> recycle.invoke(activeQueue)).get(5, TimeUnit.SECONDS); + assertTrue(event.isCommitted()); assertEquals(0L, activeQueue.getSubscriptionUncommittedEventCount()); assertEquals(0L, activeQueue.getRetainedTabletBytes()); assertFalse(ack.isDone());