This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch fix/subscription-wal-prefetch-hotspots in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit e69549b9165d300a825e0a2fd3d70eb290f67932 Author: Caideyipi <[email protected]> AuthorDate: Tue Oct 13 12:52:55 2026 +0800 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 50ae0961ce1..1407f106dae 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 class WALMetaData implements SerializedSize { } public WALMetaData(long firstSearchIndex, List<Integer> buffersSize, Set<Long> memTablesId) { + this(firstSearchIndex, buffersSize, memTablesId, 0); + } + + private WALMetaData( + long firstSearchIndex, List<Integer> buffersSize, Set<Long> 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 class WALMetaData implements SerializedSize { 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 dd5d2040009..cf12658a6e3 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 @@ public class ConsensusPrefetchingQueue { 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 @@ public class ConsensusPrefetchingQueue { 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;
