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 5c27559f02f Add subscription stall diagnostics and reduce WAL prefetch
overhead (#18808)
5c27559f02f is described below
commit 5c27559f02f525e75ee07a58d06ff8ee54578527
Author: Caideyipi <[email protected]>
AuthorDate: Mon Oct 12 14:01:18 2026 +0800
Add subscription stall diagnostics and reduce WAL prefetch overhead (#18808)
* Add subscription delivery stall and memory overcommit diagnostics
* Add debug tracing for consensus subscription prefetch rounds
* Avoid redundant WAL work during subscription prefetch
* Make WAL replay diagnostics test locale independent
* Preserve subscription WAL requests across memory backpressure
* Add opt-in subscription WAL prefetch performance comparisons
* Clarify subscription prefetch tracker synchronization
---
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 42 ++-
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 42 ++-
.../consensus/ConsensusLogToTabletConverter.java | 4 +-
.../consensus/ConsensusPrefetchingQueue.java | 314 ++++++++++++-----
.../broker/consensus/ProgressWALIterator.java | 18 +
.../consensus/SubscriptionQueueTimeTracker.java | 119 +++++++
...nsensusSubscriptionPrefetchingQueueMetrics.java | 60 ++++
.../subscription/metric/SubscriptionMetrics.java | 34 ++
.../resource/SubscriptionMemoryManager.java | 45 ++-
.../task/subtask/ConsensusPrefetchSubtask.java | 11 +
.../ConsensusLogToTabletConverterTest.java | 45 +++
...onsensusPrefetchingQueueDataNodeMemoryTest.java | 168 ++++++++-
.../consensus/ConsensusPrefetchingQueueTest.java | 79 ++++-
...nsensusPrefetchingQueueWalBackpressureTest.java | 129 +++++++
.../broker/consensus/ProgressWALIteratorTest.java | 45 ++-
.../SubscriptionWalPrefetchPerformanceTest.java | 380 +++++++++++++++++++++
.../metric/SubscriptionMetricsRestartTest.java | 87 +++++
.../resource/SubscriptionMemoryManagerTest.java | 14 +
.../iotdb/commons/service/metric/enums/Metric.java | 13 +
19 files changed, 1550 insertions(+), 99 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index e6268fbe01e..836ac88d84f 100644
---
a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -21,6 +21,46 @@ package org.apache.iotdb.db.i18n;
public final class DataNodePipeMessages {
+ public static final String
+
LOG_SUBSCRIPTION_PREFETCH_ROUND_STARTING_FOR_ARG_ACTIVE_ARG_INITIALIZED_ARG_CLOSED_ARG_PENDINGENTRIES_ARG_PREFETCHEDEVENTS_ARG_INFLIGHTEVENTS_ARG_NEXTREADSEARCHINDEX_ARG_F75FF2CD
=
+ "Subscription prefetch round starting for {}: active={},
initialized={}, closed={}, pendingEntries={}, prefetchedEvents={},
inFlightEvents={}, nextReadSearchIndex={}.";
+
+ public static final String
+
LOG_SUBSCRIPTION_PREFETCH_ROUND_COMPLETED_FOR_ARG_DURATIONMS_ARG_RESULT_ARG_DELAYMS_ARG_READSEARCHINDEX_ARG_ARG_PENDINGACCEPTED_ARG_WALACCEPTED_ARG_PENDINGENTRIES_ARG_PREFETCHEDEVENTS_ARG_INFLIGHTEVENTS_ARG_MEMORYBLOCKREASON_ARG_ADMISSIONBLOCKREASON_ARG_0F0A3D57
=
+ "Subscription prefetch round completed for {}: durationMs={},
result={}, delayMs={}, readSearchIndex={} -> {}, pendingAccepted={},
walAccepted={}, pendingEntries={}, prefetchedEvents={}, inFlightEvents={},
memoryBlockReason={}, admissionBlockReason={}.";
+
+ public static final String
+
LOG_SUBSCRIPTION_PREFETCH_ACQUIRED_QUEUE_READ_LOCK_FOR_ARG_AFTER_ARG_MS_425E9A20
=
+ "Subscription prefetch acquired queue read lock for {} after {} ms.";
+
+ public static final String
+
LOG_SUBSCRIPTION_PREFETCH_SCANNING_WAL_FOR_ARG_NEXTREADSEARCHINDEX_ARG_MAXWALENTRIES_ARG_PREFETCHEDEVENTS_ARG_AA16EEA1
=
+ "Subscription prefetch scanning WAL for {}: nextReadSearchIndex={},
maxWalEntries={}, prefetchedEvents={}.";
+
+ public static final String
+
LOG_SUBSCRIPTION_PREFETCH_PREPARING_ENTRY_FOR_ARG_SEARCHINDEX_ARG_WRITERNODEID_ARG_LOCALSEQ_ARG_EE485CC7
=
+ "Subscription prefetch preparing entry for {}: searchIndex={},
writerNodeId={}, localSeq={}.";
+
+ public static final String
+
LOG_SUBSCRIPTION_PREFETCH_PREPARED_ENTRY_FOR_ARG_SEARCHINDEX_ARG_WRITERNODEID_ARG_LOCALSEQ_ARG_TABLETS_ARG_ESTIMATEDBYTES_ARG_DURATIONMS_ARG_E91BA0CD
=
+ "Subscription prefetch prepared entry for {}: searchIndex={},
writerNodeId={}, localSeq={}, tablets={}, estimatedBytes={}, durationMs={}.";
+
+ public static final String
+
LOG_SUBSCRIPTION_PREFETCH_MEMORY_RESERVATION_REJECTED_FOR_ARG_ENTRYBYTES_ARG_REASON_ARG_QUOTABYTES_ARG_FREEBYTES_ARG_98903D3B
=
+ "Subscription prefetch memory reservation rejected for {}:
entryBytes={}, reason={}, quotaBytes={}, freeBytes={}.";
+
+ public static final String
+
LOG_SUBSCRIPTION_PREFETCH_TASK_ARG_SCHEDULING_AFTER_ROUND_RESULT_ARG_WAKEUPPENDING_ARG_ENQUEUENOW_ARG_DELAYMS_ARG_602EA4A4
=
+ "Subscription prefetch task {} scheduling after round: result={},
wakeupPending={}, enqueueNow={}, delayMs={}.";
+
+ public static final String
+
LOG_SUBSCRIPTION_MEMORY_OVERCOMMIT_ENTRYBYTES_ARG_BUDGETBYTES_ARG_OVERCOMMITBYTES_ARG_OTHER_QUEUES_MAY_BE_BLOCKED_UNTIL_RELEASE_DF9B914E
=
+ "Subscription memory overcommit: entryBytes={}, budgetBytes={},
overcommitBytes={}; other queues may be blocked until release.";
+
+ public static final String
+
LOG_SUBSCRIPTION_PREFETCH_ROUND_TOOK_ARG_MS_FOR_ARG_DELIVERYIDLETIMEMS_ARG_MEMORYUSEDBYTES_ARG_MEMORYLIMITBYTES_ARG_INSPECT_WORKER_STACKS_WAL_I_O_AND_JVM_PAUSES_D399CEDD
=
+ "Subscription prefetch round took {} ms for {}.
deliveryIdleTimeMs={}, memoryUsedBytes={}, memoryLimitBytes={}; inspect worker
stacks, WAL I/O and JVM pauses.";
+
public static final String LOG_FAILED_TO_RESOLVE_TRANSFER_EXCEPTION_A4F5397A
=
"Failed to resolve transfer exception.";
@@ -2124,7 +2164,7 @@ public final class DataNodePipeMessages {
"ConsensusPrefetchingQueue {}: periodic stats, lag={}, pendingDelta={},
walDelta={}, "
+ "pendingTotal={}, walTotal={}, walGapSkippedEntries={},
pendingQueueSize={}, "
+ "prefetchingQueueSize={}, inFlightEventsSize={},
realtimeWriterCount={}, "
- + "walHasNext={}, isActive={}, subtaskScheduled={}";
+ + "walNextBuffered={}, isActive={}, subtaskScheduled={}";
public static final String
PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAITING_MS_FOR_WAL_GAP_TO_BECOME_7D91C6C5 =
"ConsensusPrefetchingQueue {}: waiting {}ms for WAL gap [{}, {}) to
become visible, "
+ "currentNextExpected={}, currentWalIndex={}, seekGeneration={}";
diff --git
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
index 64cd9c7a2d6..5840da12daf 100644
---
a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
+++
b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java
@@ -21,6 +21,46 @@ package org.apache.iotdb.db.i18n;
public final class DataNodePipeMessages {
+ public static final String
+
LOG_SUBSCRIPTION_PREFETCH_ROUND_STARTING_FOR_ARG_ACTIVE_ARG_INITIALIZED_ARG_CLOSED_ARG_PENDINGENTRIES_ARG_PREFETCHEDEVENTS_ARG_INFLIGHTEVENTS_ARG_NEXTREADSEARCHINDEX_ARG_F75FF2CD
=
+ "订阅预取轮次开始，队列
{}：active={}，initialized={}，closed={}，pendingEntries={}，prefetchedEvents={}，inFlightEvents={}，nextReadSearchIndex={}。";
+
+ public static final String
+
LOG_SUBSCRIPTION_PREFETCH_ROUND_COMPLETED_FOR_ARG_DURATIONMS_ARG_RESULT_ARG_DELAYMS_ARG_READSEARCHINDEX_ARG_ARG_PENDINGACCEPTED_ARG_WALACCEPTED_ARG_PENDINGENTRIES_ARG_PREFETCHEDEVENTS_ARG_INFLIGHTEVENTS_ARG_MEMORYBLOCKREASON_ARG_ADMISSIONBLOCKREASON_ARG_0F0A3D57
=
+ "订阅预取轮次结束，队列
{}：durationMs={}，result={}，delayMs={}，readSearchIndex={} ->
{}，pendingAccepted={}，walAccepted={}，pendingEntries={}，prefetchedEvents={}，inFlightEvents={}，memoryBlockReason={}，admissionBlockReason={}。";
+
+ public static final String
+
LOG_SUBSCRIPTION_PREFETCH_ACQUIRED_QUEUE_READ_LOCK_FOR_ARG_AFTER_ARG_MS_425E9A20
=
+ "订阅预取已获取队列 {} 的读锁，距离轮次开始已过 {} ms。";
+
+ public static final String
+
LOG_SUBSCRIPTION_PREFETCH_SCANNING_WAL_FOR_ARG_NEXTREADSEARCHINDEX_ARG_MAXWALENTRIES_ARG_PREFETCHEDEVENTS_ARG_AA16EEA1
=
+ "订阅预取开始扫描队列 {} 的
WAL：nextReadSearchIndex={}，maxWalEntries={}，prefetchedEvents={}。";
+
+ public static final String
+
LOG_SUBSCRIPTION_PREFETCH_PREPARING_ENTRY_FOR_ARG_SEARCHINDEX_ARG_WRITERNODEID_ARG_LOCALSEQ_ARG_EE485CC7
=
+ "订阅预取开始解析和转换队列 {} 的
entry：searchIndex={}，writerNodeId={}，localSeq={}。";
+
+ public static final String
+
LOG_SUBSCRIPTION_PREFETCH_PREPARED_ENTRY_FOR_ARG_SEARCHINDEX_ARG_WRITERNODEID_ARG_LOCALSEQ_ARG_TABLETS_ARG_ESTIMATEDBYTES_ARG_DURATIONMS_ARG_E91BA0CD
=
+ "订阅预取完成队列 {} 的 entry
解析和转换：searchIndex={}，writerNodeId={}，localSeq={}，tablets={}，estimatedBytes={}，durationMs={}。";
+
+ public static final String
+
LOG_SUBSCRIPTION_PREFETCH_MEMORY_RESERVATION_REJECTED_FOR_ARG_ENTRYBYTES_ARG_REASON_ARG_QUOTABYTES_ARG_FREEBYTES_ARG_98903D3B
=
+ "订阅预取队列 {}
的内存预留被拒绝：entryBytes={}，reason={}，quotaBytes={}，freeBytes={}。";
+
+ public static final String
+
LOG_SUBSCRIPTION_PREFETCH_TASK_ARG_SCHEDULING_AFTER_ROUND_RESULT_ARG_WAKEUPPENDING_ARG_ENQUEUENOW_ARG_DELAYMS_ARG_602EA4A4
=
+ "订阅预取任务 {}
的轮次后调度：result={}，wakeupPending={}，enqueueNow={}，delayMs={}。";
+
+ public static final String
+
LOG_SUBSCRIPTION_MEMORY_OVERCOMMIT_ENTRYBYTES_ARG_BUDGETBYTES_ARG_OVERCOMMITBYTES_ARG_OTHER_QUEUES_MAY_BE_BLOCKED_UNTIL_RELEASE_DF9B914E
=
+
"订阅内存超出预算：entryBytes={}，budgetBytes={}，overcommitBytes={}；其它队列可能阻塞，直到该 entry
释放。";
+
+ public static final String
+
LOG_SUBSCRIPTION_PREFETCH_ROUND_TOOK_ARG_MS_FOR_ARG_DELIVERYIDLETIMEMS_ARG_MEMORYUSEDBYTES_ARG_MEMORYLIMITBYTES_ARG_INSPECT_WORKER_STACKS_WAL_I_O_AND_JVM_PAUSES_D399CEDD
=
+ "订阅预取轮次耗时 {} ms，队列
{}。deliveryIdleTimeMs={}，memoryUsedBytes={}，memoryLimitBytes={}；请检查工作线程栈、WAL
I/O 和 JVM 停顿。";
+
public static final String LOG_FAILED_TO_RESOLVE_TRANSFER_EXCEPTION_A4F5397A
=
"解析 transfer exception 失败。";
@@ -1970,7 +2010,7 @@ public final class DataNodePipeMessages {
public static final String
PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_PERIODIC_STATS_LAG_PENDINGDELTA_WALGAPSKIPPEDENTRIES_9A4E6608
=
"ConsensusPrefetchingQueue
{}：周期统计，lag={}，pendingDelta={}，walDelta={}，pendingTotal={}，"
+
"walTotal={}，walGapSkippedEntries={}，pendingQueueSize={}，prefetchingQueueSize={}，"
- +
"inFlightEventsSize={}，realtimeWriterCount={}，walHasNext={}，isActive={}，"
+ +
"inFlightEventsSize={}，realtimeWriterCount={}，walNextBuffered={}，isActive={}，"
+ "subtaskScheduled={}";
public static final String
PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_WAITING_MS_FOR_WAL_GAP_TO_BECOME_7D91C6C5 =
"ConsensusPrefetchingQueue {}：等待 {}ms，使 WAL 缺口 [{}, {})
可见，currentNextExpected={}，"
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java
index abd4c5cc381..882860cebbc 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverter.java
@@ -134,7 +134,9 @@ public class ConsensusLogToTabletConverter {
try {
if (req instanceof IoTConsensusRequest) {
// WAL entries read from file are wrapped as IoTConsensusRequest
(ByteBuffer).
- planNode =
WALEntry.deserializeForConsensus(req.serializeToByteBuffer());
+ // Preserve the request's buffer position so memory backpressure can
retry every
+ // fragment. duplicate() shares the bytes without retaining a second
payload copy.
+ planNode =
WALEntry.deserializeForConsensus(req.serializeToByteBuffer().duplicate());
} else if (req instanceof InsertNode) {
// In-memory entries that are not yet flushed to WAL file may
already be PlanNode.
planNode = (PlanNode) req;
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 e24748edc0b..dd5d2040009 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
@@ -319,6 +319,8 @@ public class ConsensusPrefetchingQueue {
private volatile long lastProgressTimeMs = 0L;
+ private final SubscriptionQueueTimeTracker timeTracker = new
SubscriptionQueueTimeTracker();
+
private volatile String lastConsumerId = "";
private volatile long lastPendingAcceptedEntries = 0L;
@@ -954,6 +956,7 @@ public class ConsensusPrefetchingQueue {
this.subscriptionWALIterator =
createSubscriptionWALIterator(resolvedStart.getStartSearchIndex());
this.prefetchInitialized = true;
+ timeTracker.start();
this.observedSeekGeneration = seekGeneration.get();
discardBatch(this.lingerBatch);
resetBatchWriterProgress();
@@ -1443,6 +1446,7 @@ public class ConsensusPrefetchingQueue {
// Mark as polled before updating inFlightEvents
event.recordLastPolledTimestamp();
+ timeTracker.recordDelivery();
lastPollTimeMs = System.currentTimeMillis();
lastConsumerId = consumerId;
inFlightEvents.put(new InFlightEventKey(consumerId,
event.getCommitContext()), event);
@@ -1528,7 +1532,69 @@ public class ConsensusPrefetchingQueue {
private static final long PREFETCH_STATS_LOG_INTERVAL_MS = 5_000L;
+ private static final long SLOW_PREFETCH_ROUND_THRESHOLD_MS = 10_000L;
+
+ private String getPrefetchDebugId() {
+ return getPrefetchingQueueId() + "/" + consensusGroupId;
+ }
+
public PrefetchRoundResult drivePrefetchOnce() {
+ timeTracker.beginPrefetch();
+ final boolean debugEnabled = LOGGER.isDebugEnabled();
+ final long startReadSearchIndex = debugEnabled ?
nextExpectedSearchIndex.get() : 0L;
+ final long startPendingAccepted = debugEnabled ?
pendingPathAcceptedEntries.get() : 0L;
+ final long startWalAccepted = debugEnabled ? walPathAcceptedEntries.get()
: 0L;
+ PrefetchRoundResult result = null;
+ try {
+ if (debugEnabled) {
+ LOGGER.debug(
+ DataNodePipeMessages
+
.LOG_SUBSCRIPTION_PREFETCH_ROUND_STARTING_FOR_ARG_ACTIVE_ARG_INITIALIZED_ARG_CLOSED_ARG_PENDINGENTRIES_ARG_PREFETCHEDEVENTS_ARG_INFLIGHTEVENTS_ARG_NEXTREADSEARCHINDEX_ARG_F75FF2CD,
+ getPrefetchDebugId(),
+ isActive,
+ prefetchInitialized,
+ closeRequested || isClosed,
+ pendingEntries.size(),
+ prefetchingQueue.size(),
+ inFlightEvents.size(),
+ startReadSearchIndex);
+ }
+ result = drivePrefetchOnceInternal();
+ return result;
+ } finally {
+ final long durationMs = timeTracker.endPrefetch();
+ if (debugEnabled) {
+ LOGGER.debug(
+ DataNodePipeMessages
+
.LOG_SUBSCRIPTION_PREFETCH_ROUND_COMPLETED_FOR_ARG_DURATIONMS_ARG_RESULT_ARG_DELAYMS_ARG_READSEARCHINDEX_ARG_ARG_PENDINGACCEPTED_ARG_WALACCEPTED_ARG_PENDINGENTRIES_ARG_PREFETCHEDEVENTS_ARG_INFLIGHTEVENTS_ARG_MEMORYBLOCKREASON_ARG_ADMISSIONBLOCKREASON_ARG_0F0A3D57,
+ getPrefetchDebugId(),
+ durationMs,
+ Objects.nonNull(result) ? result.getType() : null,
+ Objects.nonNull(result) ? result.getDelayMs() : null,
+ startReadSearchIndex,
+ nextExpectedSearchIndex.get(),
+ pendingPathAcceptedEntries.get() - startPendingAccepted,
+ walPathAcceptedEntries.get() - startWalAccepted,
+ pendingEntries.size(),
+ prefetchingQueue.size(),
+ inFlightEvents.size(),
+ memoryBlockReason,
+ realtimeAdmissionBlockReason);
+ }
+ if (durationMs >= SLOW_PREFETCH_ROUND_THRESHOLD_MS) {
+ LOGGER.warn(
+ DataNodePipeMessages
+
.LOG_SUBSCRIPTION_PREFETCH_ROUND_TOOK_ARG_MS_FOR_ARG_DELIVERYIDLETIMEMS_ARG_MEMORYUSEDBYTES_ARG_MEMORYLIMITBYTES_ARG_INSPECT_WORKER_STACKS_WAL_I_O_AND_JVM_PAUSES_D399CEDD,
+ durationMs,
+ this,
+ getDeliveryIdleTimeMs(),
+ subscriptionMemoryManager.getUsedMemorySizeInBytes(),
+ subscriptionMemoryManager.getTotalMemorySizeInBytes());
+ }
+ }
+ }
+
+ private PrefetchRoundResult drivePrefetchOnceInternal() {
if (applyPendingSeekRequestIfNecessary()) {
return closeRequested ? PrefetchRoundResult.dormant() :
PrefetchRoundResult.rescheduleNow();
}
@@ -1539,6 +1605,13 @@ public class ConsensusPrefetchingQueue {
return PrefetchRoundResult.dormant();
}
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug(
+ DataNodePipeMessages
+
.LOG_SUBSCRIPTION_PREFETCH_ACQUIRED_QUEUE_READ_LOCK_FOR_ARG_AFTER_ARG_MS_425E9A20,
+ getPrefetchDebugId(),
+ getPrefetchDurationMs());
+ }
logPeriodicStatsIfNecessary();
final long currentSeekGeneration = seekGeneration.get();
@@ -1714,7 +1787,7 @@ public class ConsensusPrefetchingQueue {
prefetchingQueue.size(),
inFlightEvents.size(),
realtimeEntriesByWriter.size(),
- hasReadableWalEntries(),
+ Objects.nonNull(subscriptionWALIterator) &&
subscriptionWALIterator.hasBufferedNext(),
isActive,
Objects.nonNull(prefetchSubtask) &&
prefetchSubtask.isScheduledOrRunning());
lastStatsLogTimeMs = nowMs;
@@ -1819,24 +1892,9 @@ public class ConsensusPrefetchingQueue {
return hasLocalSearchIndex(request) && request.getSearchIndex() <
nextExpectedSearchIndex.get();
}
- private boolean advanceLocalCursorIfPresent(final IndexedConsensusRequest
request) {
+ private void advanceLocalCursorIfPresent(final IndexedConsensusRequest
request) {
if (hasLocalSearchIndex(request)) {
nextExpectedSearchIndex.set(request.getSearchIndex() + 1);
- return true;
- }
- return false;
- }
-
- private void advanceLocalCursorFromPendingIfPresent(
- final IndexedConsensusRequest request, final long
expectedSeekGeneration) {
- if (advanceLocalCursorIfPresent(request)) {
- // Pending delivery advances independently of the WAL reader. Raise its
local lower bound in
- // place so stale local requests are filtered without rebuilding and
rescanning retained WAL.
- final ProgressWALIterator iterator = subscriptionWALIterator;
- if (Objects.nonNull(iterator) && seekGeneration.get() ==
expectedSeekGeneration) {
- iterator.advanceTo(
- nextExpectedSearchIndex.get(),
this::isWriterProgressCoveredForWalFastForward);
- }
}
}
@@ -1895,76 +1953,90 @@ public class ConsensusPrefetchingQueue {
int processedCount = 0;
int skippedCount = 0;
+ final long startSearchIndex = nextExpectedSearchIndex.get();
- for (int index = 0; index < batch.size(); index++) {
- final IndexedConsensusRequest request = batch.get(index);
- final long searchIndex = request.getSearchIndex();
+ try {
+ for (int index = 0; index < batch.size(); index++) {
+ final IndexedConsensusRequest request = batch.get(index);
+ final long searchIndex = request.getSearchIndex();
- // Only local-indexed requests participate in the internal WAL read
cursor.
- final long expected = nextExpectedSearchIndex.get();
- if (hasLocalSearchIndex(request) && searchIndex > expected) {
- LOGGER.debug(
- DataNodePipeMessages
-
.PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_GAP_DETECTED_EXPECTED_GOT_FILLING_70DD08B3,
- this,
- expected,
- searchIndex,
- searchIndex - expected);
- final MaterializationResult gapFillResult =
- fillGapFromWAL(
- expected,
- searchIndex,
- lingerBatch,
- expectedSeekGeneration,
- maxWalEntries,
- maxTablets,
- maxBatchBytes);
- if (gapFillResult != MaterializationResult.SUCCESS) {
- return gapFillResult;
+ // Only local-indexed requests participate in the internal WAL read
cursor.
+ final long expected = nextExpectedSearchIndex.get();
+ if (hasLocalSearchIndex(request) && searchIndex > expected) {
+ LOGGER.debug(
+ DataNodePipeMessages
+
.PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_GAP_DETECTED_EXPECTED_GOT_FILLING_70DD08B3,
+ this,
+ expected,
+ searchIndex,
+ searchIndex - expected);
+ final MaterializationResult gapFillResult =
+ fillGapFromWAL(
+ expected,
+ searchIndex,
+ lingerBatch,
+ expectedSeekGeneration,
+ maxWalEntries,
+ maxTablets,
+ maxBatchBytes);
+ if (gapFillResult != MaterializationResult.SUCCESS) {
+ return gapFillResult;
+ }
}
- }
- if (isBeforeLocalCursor(request)) {
- skippedCount++;
- continue;
- }
+ if (isBeforeLocalCursor(request)) {
+ skippedCount++;
+ continue;
+ }
- if (shouldSkipForRecoveryProgress(request)) {
- skippedCount++;
- advanceLocalCursorFromPendingIfPresent(request,
expectedSeekGeneration);
- continue;
- }
- if (shouldSkipForMaterializedProgress(request)) {
- skippedCount++;
- advanceLocalCursorFromPendingIfPresent(request,
expectedSeekGeneration);
- continue;
- }
+ if (shouldSkipForRecoveryProgress(request)) {
+ skippedCount++;
+ advanceLocalCursorIfPresent(request);
+ continue;
+ }
+ if (shouldSkipForMaterializedProgress(request)) {
+ skippedCount++;
+ advanceLocalCursorIfPresent(request);
+ continue;
+ }
- final MaterializationResult appendResult =
- appendRealtimeRequest(
- request, lingerBatch, expectedSeekGeneration, maxTablets,
maxBatchBytes, true);
- if (appendResult != MaterializationResult.SUCCESS) {
- return appendResult;
- }
- markMaterializedProgress(request);
- processedCount++;
- advanceLocalCursorFromPendingIfPresent(request, expectedSeekGeneration);
- if (prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE) {
- break;
+ final MaterializationResult appendResult =
+ appendRealtimeRequest(
+ request, lingerBatch, expectedSeekGeneration, maxTablets,
maxBatchBytes, true);
+ if (appendResult != MaterializationResult.SUCCESS) {
+ return appendResult;
+ }
+ markMaterializedProgress(request);
+ processedCount++;
+ advanceLocalCursorIfPresent(request);
+ if (prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE) {
+ break;
+ }
}
- }
- LOGGER.debug(
- DataNodePipeMessages
-
.PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_ACCUMULATE_COMPLETE_BATCHSIZE_FA3F3B41,
- this,
- batch.size(),
- processedCount,
- skippedCount,
- lingerBatch.tablets.size(),
- nextExpectedSearchIndex.get());
+ LOGGER.debug(
+ DataNodePipeMessages
+
.PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_ACCUMULATE_COMPLETE_BATCHSIZE_FA3F3B41,
+ this,
+ batch.size(),
+ processedCount,
+ skippedCount,
+ lingerBatch.tablets.size(),
+ nextExpectedSearchIndex.get());
- return MaterializationResult.SUCCESS;
+ return MaterializationResult.SUCCESS;
+ } finally {
+ // Advance once for the consumed prefix, including when a gap or memory
block stops the
+ // batch. Per-entry fast-forward can repeatedly inspect the same
uncovered WAL file footer.
+ // Gap recovery creates its own iterator at the current cursor before
reading WAL.
+ final ProgressWALIterator iterator = subscriptionWALIterator;
+ if (nextExpectedSearchIndex.get() > startSearchIndex
+ && Objects.nonNull(iterator)
+ && seekGeneration.get() == expectedSeekGeneration) {
+ iterator.advanceTo(
+ nextExpectedSearchIndex.get(),
this::isWriterProgressCoveredForWalFastForward);
+ }
+ }
}
/**
@@ -2080,6 +2152,15 @@ public class ConsensusPrefetchingQueue {
return MaterializationResult.SUCCESS;
}
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug(
+ DataNodePipeMessages
+
.LOG_SUBSCRIPTION_PREFETCH_SCANNING_WAL_FOR_ARG_NEXTREADSEARCHINDEX_ARG_MAXWALENTRIES_ARG_PREFETCHEDEVENTS_ARG_AA16EEA1,
+ getPrefetchDebugId(),
+ nextExpectedSearchIndex.get(),
+ maxWalEntries,
+ prefetchingQueue.size());
+ }
ensureSubscriptionWalReadable();
reportUnreadableWalReplayIfNecessary();
@@ -2109,7 +2190,10 @@ public class ConsensusPrefetchingQueue {
walEntry, batchState, expectedSeekGeneration, maxTablets,
maxBatchBytes, false);
if (appendResult != MaterializationResult.SUCCESS) {
if (appendResult == MaterializationResult.MEMORY_BLOCKED) {
- resetSubscriptionWALPosition(nextExpectedSearchIndex.get());
+ // next() has removed this request from the iterator, but replay
progress has not
+ // advanced. Keep it buffered so the next round retries it without
reopening the WAL
+ // and rescanning the prefix before the current reader position.
+ subscriptionWALIterator.retry(walEntry);
}
return appendResult;
}
@@ -2304,6 +2388,17 @@ public class ConsensusPrefetchingQueue {
}
private PreparedEntry prepareEntry(final IndexedConsensusRequest
indexedRequest) {
+ final boolean debugEnabled = LOGGER.isDebugEnabled();
+ final long prepareStartNs = debugEnabled ? System.nanoTime() : 0L;
+ if (debugEnabled) {
+ LOGGER.debug(
+ DataNodePipeMessages
+
.LOG_SUBSCRIPTION_PREFETCH_PREPARING_ENTRY_FOR_ARG_SEARCHINDEX_ARG_WRITERNODEID_ARG_LOCALSEQ_ARG_EE485CC7,
+ getPrefetchDebugId(),
+ indexedRequest.getSearchIndex(),
+ indexedRequest.getNodeId(),
+ indexedRequest.getProgressLocalSeq());
+ }
final InsertNode insertNode =
ConsensusLogToTabletConverter.deserializeToInsertNode(indexedRequest);
if (Objects.isNull(insertNode)) {
@@ -2330,6 +2425,18 @@ public class ConsensusPrefetchingQueue {
}
final List<Tablet> tablets = converter.convert(insertNode);
final long estimatedBytes = estimateTabletsBytes(tablets);
+ if (debugEnabled) {
+ LOGGER.debug(
+ DataNodePipeMessages
+
.LOG_SUBSCRIPTION_PREFETCH_PREPARED_ENTRY_FOR_ARG_SEARCHINDEX_ARG_WRITERNODEID_ARG_LOCALSEQ_ARG_TABLETS_ARG_ESTIMATEDBYTES_ARG_DURATIONMS_ARG_E91BA0CD,
+ getPrefetchDebugId(),
+ searchIndex,
+ writerNodeId,
+ localSeq,
+ tablets.size(),
+ estimatedBytes,
+ TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - prepareStartNs));
+ }
if (!tryReserveTabletMemory(estimatedBytes)) {
return null;
}
@@ -2472,6 +2579,16 @@ public class ConsensusPrefetchingQueue {
final SubscriptionMemoryManager.AllocationResult allocationResult =
subscriptionMemoryHandle.tryAllocate(bytes);
if (!allocationResult.isAccepted()) {
+ if (LOGGER.isDebugEnabled()) {
+ LOGGER.debug(
+ DataNodePipeMessages
+
.LOG_SUBSCRIPTION_PREFETCH_MEMORY_RESERVATION_REJECTED_FOR_ARG_ENTRYBYTES_ARG_REASON_ARG_QUOTABYTES_ARG_FREEBYTES_ARG_98903D3B,
+ getPrefetchDebugId(),
+ bytes,
+ allocationResult.getRejectionReason(),
+ allocationResult.getQuotaInBytes(),
+ allocationResult.getFreeMemoryInBytes());
+ }
memoryBlockedEntryBytes = bytes;
final SubscriptionQueueRejectionReason rejectionReason =
toQueueRejectionReason(allocationResult.getRejectionReason());
@@ -4378,6 +4495,30 @@ public class ConsensusPrefetchingQueue {
return lastProgressTimeMs;
}
+ public long getLastDeliveryIntervalMs() {
+ return timeTracker.getLastDeliveryIntervalMs();
+ }
+
+ public long getMaxDeliveryIntervalMs() {
+ return timeTracker.getMaxDeliveryIntervalMs();
+ }
+
+ public long getDeliveryIdleTimeMs() {
+ return timeTracker.getDeliveryIdleTimeMs();
+ }
+
+ public long getPrefetchDurationMs() {
+ return timeTracker.getPrefetchDurationMs();
+ }
+
+ public long getMaxPrefetchDurationMs() {
+ return timeTracker.getMaxPrefetchDurationMs();
+ }
+
+ public long getPrefetchIdleTimeMs() {
+ return timeTracker.getPrefetchIdleTimeMs();
+ }
+
/** Returns 0=uninitialized, 1=inactive, 2=caught up, 3=catching up,
4=stalled. */
public long getProgressStatus() {
switch (getProgressStatusName()) {
@@ -4743,6 +4884,21 @@ public class ConsensusPrefetchingQueue {
result.put("retainedRequestBytes",
String.valueOf(getRetainedRequestBytes()));
result.put("retainedTabletBytes",
String.valueOf(retainedTabletBytes.get()));
result.put("memoryBlockedEntryBytes",
String.valueOf(memoryBlockedEntryBytes));
+ result.put(
+ "dataNodeSubscriptionMemoryUsedInBytes",
+ String.valueOf(subscriptionMemoryManager.getUsedMemorySizeInBytes()));
+ result.put(
+ "subscriptionMemoryOvercommitBytes",
+ String.valueOf(subscriptionMemoryManager.getOvercommitSizeInBytes()));
+ result.put(
+ "subscriptionMemoryOversizedEntryCount",
+ String.valueOf(subscriptionMemoryManager.getOversizedEntryCount()));
+ result.put("lastDeliveryIntervalMs",
String.valueOf(getLastDeliveryIntervalMs()));
+ result.put("maxDeliveryIntervalMs",
String.valueOf(getMaxDeliveryIntervalMs()));
+ result.put("deliveryIdleTimeMs", String.valueOf(getDeliveryIdleTimeMs()));
+ result.put("prefetchDurationMs", String.valueOf(getPrefetchDurationMs()));
+ result.put("maxPrefetchDurationMs",
String.valueOf(getMaxPrefetchDurationMs()));
+ result.put("prefetchIdleTimeMs", String.valueOf(getPrefetchIdleTimeMs()));
result.put("realtimeAdmissionBlocked",
String.valueOf(realtimeAdmissionBlocked.get()));
result.put(
"subscriptionMemoryLimitInBytes",
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java
index cbf0649f224..3636829274f 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIterator.java
@@ -49,6 +49,7 @@ import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
+import java.util.Objects;
import java.util.Set;
/**
@@ -205,6 +206,23 @@ public class ProgressWALIterator implements Closeable,
Iterator<IndexedConsensus
}
}
+ /**
+ * Returns whether a usable request is already buffered, without reading or
advancing WAL. A false
+ * result does not imply that the iterator is exhausted.
+ */
+ boolean hasBufferedNext() {
+ return nextReady != null && !shouldSkip(nextReady);
+ }
+
+ /**
+ * Keeps a request returned by {@link #next()} available for a retry without
reopening the WAL.
+ * The caller uses this when materialization is temporarily blocked and
replay progress has not
+ * advanced yet.
+ */
+ void retry(final IndexedConsensusRequest request) {
+ nextReady = Objects.requireNonNull(request);
+ }
+
/**
* Advances the local search-index lower bound without rebuilding the
iterator. Whole WAL files
* are skipped only when every writer progress tuple in them is already
covered by queue state.
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/SubscriptionQueueTimeTracker.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/SubscriptionQueueTimeTracker.java
new file mode 100644
index 00000000000..0a8187e4e2e
--- /dev/null
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/SubscriptionQueueTimeTracker.java
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.subscription.broker.consensus;
+
+import org.apache.iotdb.commons.utils.TestOnly;
+
+import java.util.concurrent.TimeUnit;
+import java.util.function.LongSupplier;
+
+/** Monotonic timing that remains readable while a queue's prefetch worker is
blocked. */
+class SubscriptionQueueTimeTracker {
+
+ private final LongSupplier nanoTime;
+
+ private volatile boolean started;
+ private boolean delivered;
+ private volatile long lastDeliveryTimeNs;
+ private volatile long lastDeliveryIntervalMs;
+ private volatile long maxDeliveryIntervalMs;
+ private volatile long lastPrefetchCompletionTimeNs;
+ private volatile long prefetchStartTimeNs;
+ private volatile boolean prefetchRunning;
+ private volatile long maxPrefetchDurationMs;
+
+ SubscriptionQueueTimeTracker() {
+ this(System::nanoTime);
+ }
+
+ @TestOnly
+ SubscriptionQueueTimeTracker(final LongSupplier nanoTime) {
+ this.nanoTime = nanoTime;
+ }
+
+ synchronized void start() {
+ if (!started) {
+ final long nowNs = nanoTime.getAsLong();
+ lastDeliveryTimeNs = nowNs;
+ lastPrefetchCompletionTimeNs = nowNs;
+ started = true;
+ }
+ }
+
+ // Multiple consumers may poll the queue concurrently under its read lock,
so delivery interval
+ // updates must remain atomic even though prefetch rounds are serialized.
+ synchronized void recordDelivery() {
+ start();
+ final long nowNs = nanoTime.getAsLong();
+ if (delivered) {
+ lastDeliveryIntervalMs = elapsedMs(nowNs, lastDeliveryTimeNs);
+ maxDeliveryIntervalMs = Math.max(maxDeliveryIntervalMs,
lastDeliveryIntervalMs);
+ }
+ lastDeliveryTimeNs = nowNs;
+ delivered = true;
+ }
+
+ // The prefetch subtask serializes rounds, so this path does not need
synchronization beyond the
+ // one-time initialization performed by start().
+ void beginPrefetch() {
+ start();
+ prefetchStartTimeNs = nanoTime.getAsLong();
+ prefetchRunning = true;
+ }
+
+ long endPrefetch() {
+ final long nowNs = nanoTime.getAsLong();
+ final long durationMs = elapsedMs(nowNs, prefetchStartTimeNs);
+ maxPrefetchDurationMs = Math.max(maxPrefetchDurationMs, durationMs);
+ lastPrefetchCompletionTimeNs = nowNs;
+ prefetchRunning = false;
+ return durationMs;
+ }
+
+ long getLastDeliveryIntervalMs() {
+ return lastDeliveryIntervalMs;
+ }
+
+ long getMaxDeliveryIntervalMs() {
+ return maxDeliveryIntervalMs;
+ }
+
+ long getDeliveryIdleTimeMs() {
+ return started ? elapsedMs(nanoTime.getAsLong(), lastDeliveryTimeNs) : 0L;
+ }
+
+ /** Elapsed time of an ongoing round, including waiting to acquire the queue
lock. */
+ long getPrefetchDurationMs() {
+ return prefetchRunning ? elapsedMs(nanoTime.getAsLong(),
prefetchStartTimeNs) : 0L;
+ }
+
+ long getMaxPrefetchDurationMs() {
+ return Math.max(maxPrefetchDurationMs, getPrefetchDurationMs());
+ }
+
+ /** Time since the last completed round, or activation before the first
completed round. */
+ long getPrefetchIdleTimeMs() {
+ return started ? elapsedMs(nanoTime.getAsLong(),
lastPrefetchCompletionTimeNs) : 0L;
+ }
+
+ private static long elapsedMs(final long nowNs, final long thenNs) {
+ return Math.max(0L, TimeUnit.NANOSECONDS.toMillis(nowNs - thenNs));
+ }
+}
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java
index 359da8f9239..f0c3cfed01d 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetrics.java
@@ -206,6 +206,42 @@ public class ConsensusSubscriptionPrefetchingQueueMetrics
implements IMetricSet
queue,
ConsensusPrefetchingQueue::getLastProgressTimeMs,
key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_LAST_DELIVERY_INTERVAL_MS.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getLastDeliveryIntervalMs,
+ key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_MAX_DELIVERY_INTERVAL_MS.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getMaxDeliveryIntervalMs,
+ key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_DELIVERY_IDLE_TIME_MS.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getDeliveryIdleTimeMs,
+ key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_PREFETCH_DURATION_MS.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getPrefetchDurationMs,
+ key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_MAX_PREFETCH_DURATION_MS.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getMaxPrefetchDurationMs,
+ key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_PREFETCH_IDLE_TIME_MS.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getPrefetchIdleTimeMs,
+ key.getTags());
metricService.createAutoGauge(
Metric.SUBSCRIPTION_CONSENSUS_PROGRESS_STATUS.toString(),
MetricLevel.IMPORTANT,
@@ -373,6 +409,30 @@ public class ConsensusSubscriptionPrefetchingQueueMetrics
implements IMetricSet
MetricType.AUTO_GAUGE,
Metric.SUBSCRIPTION_CONSENSUS_LAST_PROGRESS_TIME.toString(),
key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.SUBSCRIPTION_CONSENSUS_LAST_DELIVERY_INTERVAL_MS.toString(),
+ key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.SUBSCRIPTION_CONSENSUS_MAX_DELIVERY_INTERVAL_MS.toString(),
+ key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.SUBSCRIPTION_CONSENSUS_DELIVERY_IDLE_TIME_MS.toString(),
+ key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.SUBSCRIPTION_CONSENSUS_PREFETCH_DURATION_MS.toString(),
+ key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.SUBSCRIPTION_CONSENSUS_MAX_PREFETCH_DURATION_MS.toString(),
+ key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.SUBSCRIPTION_CONSENSUS_PREFETCH_IDLE_TIME_MS.toString(),
+ key.getTags());
metricService.remove(
MetricType.AUTO_GAUGE,
Metric.SUBSCRIPTION_CONSENSUS_PROGRESS_STATUS.toString(),
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/SubscriptionMetrics.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/SubscriptionMetrics.java
index 29de59ddf32..8e8657bb3d5 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/SubscriptionMetrics.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/metric/SubscriptionMetrics.java
@@ -19,8 +19,13 @@
package org.apache.iotdb.db.subscription.metric;
+import org.apache.iotdb.commons.service.metric.enums.Metric;
+import
org.apache.iotdb.db.subscription.resource.SubscriptionDataNodeResourceManager;
+import org.apache.iotdb.db.subscription.resource.SubscriptionMemoryManager;
import org.apache.iotdb.metrics.AbstractMetricService;
import org.apache.iotdb.metrics.metricsets.IMetricSet;
+import org.apache.iotdb.metrics.utils.MetricLevel;
+import org.apache.iotdb.metrics.utils.MetricType;
public class SubscriptionMetrics implements IMetricSet {
@@ -30,12 +35,41 @@ public class SubscriptionMetrics implements IMetricSet {
public void bindTo(final AbstractMetricService metricService) {
SubscriptionPrefetchingQueueMetrics.getInstance().bindTo(metricService);
ConsensusSubscriptionPrefetchingQueueMetrics.getInstance().bindTo(metricService);
+ // Read the shared allocator directly; queue snapshots taken at different
times cannot be
+ // summed to establish whether a DataNode exceeded its subscription memory
budget.
+ final SubscriptionMemoryManager memoryManager =
SubscriptionDataNodeResourceManager.memory();
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_MEMORY_USED_BYTES.toString(),
+ MetricLevel.IMPORTANT,
+ memoryManager,
+ SubscriptionMemoryManager::getUsedMemorySizeInBytes);
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_MEMORY_LIMIT_BYTES.toString(),
+ MetricLevel.IMPORTANT,
+ memoryManager,
+ SubscriptionMemoryManager::getTotalMemorySizeInBytes);
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_MEMORY_OVERCOMMIT_BYTES.toString(),
+ MetricLevel.IMPORTANT,
+ memoryManager,
+ SubscriptionMemoryManager::getOvercommitSizeInBytes);
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_MEMORY_OVERSIZED_ENTRY_COUNT.toString(),
+ MetricLevel.IMPORTANT,
+ memoryManager,
+ SubscriptionMemoryManager::getOversizedEntryCount);
}
@Override
public void unbindFrom(final AbstractMetricService metricService) {
SubscriptionPrefetchingQueueMetrics.getInstance().unbindFrom(metricService);
ConsensusSubscriptionPrefetchingQueueMetrics.getInstance().unbindFrom(metricService);
+ metricService.remove(MetricType.AUTO_GAUGE,
Metric.SUBSCRIPTION_MEMORY_USED_BYTES.toString());
+ metricService.remove(MetricType.AUTO_GAUGE,
Metric.SUBSCRIPTION_MEMORY_LIMIT_BYTES.toString());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
Metric.SUBSCRIPTION_MEMORY_OVERCOMMIT_BYTES.toString());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
Metric.SUBSCRIPTION_MEMORY_OVERSIZED_ENTRY_COUNT.toString());
}
//////////////////////////// singleton ////////////////////////////
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/resource/SubscriptionMemoryManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/resource/SubscriptionMemoryManager.java
index d250cd0ba16..f13ef2b1329 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/resource/SubscriptionMemoryManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/resource/SubscriptionMemoryManager.java
@@ -24,9 +24,14 @@ import org.apache.iotdb.commons.memory.IMemoryBlock;
import org.apache.iotdb.commons.memory.MemoryBlockType;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.i18n.DataNodePipeMessages;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
/**
* DataNode-wide memory manager for materialized subscription data.
@@ -39,8 +44,12 @@ import java.util.Map;
*/
public class SubscriptionMemoryManager {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(SubscriptionMemoryManager.class);
+
private static final String MEMORY_BLOCK_NAME = "Subscription";
+ private static final long OVERCOMMIT_WARNING_INTERVAL_NS =
TimeUnit.SECONDS.toNanos(30);
+
private final IMemoryBlock memoryBlock;
private static final int PROTECTED_SHARE_DIVISOR = 2;
private final Map<Long, MemoryHandle> handles = new HashMap<>();
@@ -49,6 +58,10 @@ public class SubscriptionMemoryManager {
// Kept for callers of the original manager-level API. It is deliberately
not a fair-share owner.
private long legacyUsedMemoryInBytes;
+ private volatile long oversizedEntryCount = 0L;
+
+ private long lastOvercommitWarningTimeNs;
+
SubscriptionMemoryManager() {
memoryBlock =
IoTDBDescriptor.getInstance()
@@ -71,7 +84,17 @@ public class SubscriptionMemoryManager {
return handle;
}
- /** Reserves memory through the compatibility owner. New queues should use a
handle. */
+ /**
+ * Reserves memory through the compatibility owner. Consensus queues use a
handle, which rejects
+ * oversized entries instead of overcommitting the node budget.
+ *
+ * <p>A single entry larger than the whole budget is allowed only while the
budget is otherwise
+ * empty. This avoids permanently blocking progress while keeping the
overrun bounded by one
+ * entry.
+ *
+ * <p>This is a soft limit for such entries. Other queues can be blocked
until the oversized entry
+ * is released; the exception is exposed through metrics and a rate-limited
warning.
+ */
public synchronized boolean tryAllocate(final long sizeInBytes) {
return tryAllocateLegacy(sizeInBytes).isAccepted();
}
@@ -93,6 +116,18 @@ public class SubscriptionMemoryManager {
if (total > 0L && used == 0L && sizeInBytes > total) {
memoryBlock.forceAllocateWithoutLimitation(sizeInBytes);
legacyUsedMemoryInBytes += sizeInBytes;
+ oversizedEntryCount++;
+ final long nowNs = System.nanoTime();
+ if (oversizedEntryCount == 1L
+ || nowNs - lastOvercommitWarningTimeNs >=
OVERCOMMIT_WARNING_INTERVAL_NS) {
+ lastOvercommitWarningTimeNs = nowNs;
+ LOGGER.warn(
+ DataNodePipeMessages
+
.LOG_SUBSCRIPTION_MEMORY_OVERCOMMIT_ENTRYBYTES_ARG_BUDGETBYTES_ARG_OVERCOMMITBYTES_ARG_OTHER_QUEUES_MAY_BE_BLOCKED_UNTIL_RELEASE_DF9B914E,
+ sizeInBytes,
+ memoryBlock.getTotalMemorySizeInBytes(),
+ getOvercommitSizeInBytes());
+ }
return AllocationResult.accepted(total, total - sizeInBytes);
}
return AllocationResult.rejected(
@@ -126,6 +161,14 @@ public class SubscriptionMemoryManager {
return memoryBlock.getFreeMemoryInBytes();
}
+ public synchronized long getOvercommitSizeInBytes() {
+ return Math.max(0L, getUsedMemorySizeInBytes() -
getTotalMemorySizeInBytes());
+ }
+
+ public long getOversizedEntryCount() {
+ return oversizedEntryCount;
+ }
+
private synchronized AllocationResult tryAllocate(
final MemoryHandle handle, final long sizeInBytes) {
final AllocationResult decision = inspectAllocationUnderLock(handle,
sizeInBytes);
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/ConsensusPrefetchSubtask.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/ConsensusPrefetchSubtask.java
index 703289b736b..5f91fecd3d1 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/ConsensusPrefetchSubtask.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/task/subtask/ConsensusPrefetchSubtask.java
@@ -124,6 +124,7 @@ public class ConsensusPrefetchSubtask {
boolean shouldEnqueue = false;
Long delayedWakeMs = null;
long delayedToken = 0L;
+ boolean wakeupRequested;
synchronized (monitor) {
running = false;
if (closed) {
@@ -132,6 +133,7 @@ public class ConsensusPrefetchSubtask {
return;
}
+ wakeupRequested = wakeupPending;
if (wakeupPending) {
wakeupPending = false;
shouldEnqueue = true;
@@ -158,6 +160,15 @@ public class ConsensusPrefetchSubtask {
monitor.notifyAll();
}
+ LOGGER.debug(
+ DataNodePipeMessages
+
.LOG_SUBSCRIPTION_PREFETCH_TASK_ARG_SCHEDULING_AFTER_ROUND_RESULT_ARG_WAKEUPPENDING_ARG_ENQUEUENOW_ARG_DELAYMS_ARG_602EA4A4,
+ taskId,
+ result.getType(),
+ wakeupRequested,
+ shouldEnqueue,
+ delayedWakeMs);
+
final ConsensusSubscriptionPrefetchExecutor currentExecutor = executor;
if (currentExecutor == null) {
return;
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java
index b06ccf453fe..c4af27f764c 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusLogToTabletConverterTest.java
@@ -24,7 +24,11 @@ import org.apache.iotdb.commons.path.PartialPath;
import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern;
import org.apache.iotdb.commons.pipe.datastructure.pattern.TablePattern;
import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId;
+import org.apache.iotdb.commons.request.IConsensusRequest;
import org.apache.iotdb.commons.schema.table.column.TsTableColumnCategory;
+import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest;
+import org.apache.iotdb.consensus.common.request.IoTConsensusRequest;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsOfOneDeviceNode;
@@ -32,6 +36,8 @@ import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTablet
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode;
import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertTabletNode;
import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils;
+import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALInfoEntry;
+import
org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest;
import org.apache.iotdb.db.subscription.columnfilter.ColumnFilterMatcher;
import org.apache.tsfile.enums.ColumnCategory;
@@ -44,6 +50,7 @@ import org.apache.tsfile.write.schema.MeasurementSchema;
import org.junit.Assert;
import org.junit.Test;
+import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.time.LocalDate;
import java.util.ArrayList;
@@ -55,6 +62,44 @@ public class ConsensusLogToTabletConverterTest {
private static final String DATABASE_NAME = "db";
+ @Test
+ public void testRepeatedWalDeserializationPreservesAllFragments() {
+ final List<IConsensusRequest> fragments = new ArrayList<>();
+ for (int row = 1; row <= 2; row++) {
+ final RelationalInsertRowNode node =
StatementTestUtils.genInsertRowNode(row);
+ node.setSearchIndex(4L);
+ node.setLastFragment(row == 2);
+ final WALInfoEntry walEntry = new WALInfoEntry(1L, node);
+ final ByteBuffer buffer = ByteBuffer.allocate(walEntry.serializedSize());
+ walEntry.serialize(new WALByteBufferForTest(buffer));
+ buffer.flip();
+ fragments.add(new IoTConsensusRequest(buffer));
+ }
+ final IndexedConsensusRequest request =
+ new IndexedConsensusRequest(4L, 4L,
fragments).setPhysicalTime(1000L).setNodeId(7);
+ final ConsensusLogToTabletConverter converter = createConverter("id1",
"m1");
+
+ for (int attempt = 0; attempt < 3; attempt++) {
+ final InsertNode node =
ConsensusLogToTabletConverter.deserializeToInsertNode(request);
+ Assert.assertNotNull(node);
+ Assert.assertEquals(4L, node.getSearchIndex());
+ Assert.assertEquals(4L, node.getSyncIndex());
+ Assert.assertEquals(1000L, node.getPhysicalTime());
+ Assert.assertEquals(7, node.getNodeId());
+ final List<Tablet> tablets = converter.convert(node);
+ Assert.assertEquals(2, tablets.size());
+ for (int row = 1; row <= 2; row++) {
+ final Tablet tablet = tablets.get(row - 1);
+ Assert.assertEquals(1, tablet.getRowSize());
+ Assert.assertEquals("id:" + row, toUtf8(((Binary[])
tablet.getValues()[0])[0]));
+ Assert.assertEquals((double) row, ((double[])
tablet.getValues()[1])[0], 0.0);
+ final ByteBuffer buffer = fragments.get(row -
1).serializeToByteBuffer();
+ Assert.assertEquals(0, buffer.position());
+ Assert.assertEquals(buffer.capacity(), buffer.limit());
+ }
+ }
+ }
+
@Test
public void testDataModelIsDerivedFromImmutableConversionPattern() {
final ConsensusLogToTabletConverter tableConverter =
createConverter("id1");
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDataNodeMemoryTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDataNodeMemoryTest.java
index c7ebf06ab05..16f675a4da2 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDataNodeMemoryTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueDataNodeMemoryTest.java
@@ -36,6 +36,7 @@ import
org.apache.iotdb.db.subscription.resource.SubscriptionMemoryManager;
import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.apache.iotdb.rpc.subscription.payload.poll.ErrorPayload;
import org.apache.iotdb.rpc.subscription.payload.poll.RegionProgress;
+import
org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionPollResponseType;
import org.apache.tsfile.enums.ColumnCategory;
import org.apache.tsfile.enums.TSDataType;
@@ -48,12 +49,19 @@ import org.junit.rules.TemporaryFolder;
import java.io.File;
import java.lang.reflect.Constructor;
import java.lang.reflect.Field;
+import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -226,6 +234,12 @@ public class ConsensusPrefetchingQueueDataNodeMemoryTest {
assertEquals(1L, largeQueue.getOversizedEntryRejectionCount());
assertEquals(0L, largeQueue.getRealtimeAdmissionRejectionCount());
assertEquals(smallBytes, memoryManager.getUsedMemorySizeInBytes());
+ assertEquals(0L, memoryManager.getOvercommitSizeInBytes());
+ assertEquals(0L, memoryManager.getOversizedEntryCount());
+ assertEquals("0",
largeQueue.coreReportMessage().get("subscriptionMemoryUsedInBytes"));
+ assertEquals(
+ String.valueOf(smallBytes),
+
largeQueue.coreReportMessage().get("dataNodeSubscriptionMemoryUsedInBytes"));
final SubscriptionEvent error = largeQueue.poll("largeConsumer");
assertNotNull(error);
final ErrorPayload payload = (ErrorPayload)
error.getCurrentResponse().getPayload();
@@ -306,9 +320,10 @@ public class ConsensusPrefetchingQueueDataNodeMemoryTest {
assertEquals(oneTabletBytes, queueA.getRetainedTabletBytes());
assertEquals(0L, queueB.getRetainedTabletBytes());
assertEquals(oneTabletBytes, memoryManager.getUsedMemorySizeInBytes());
- assertTrue(
- queueA.getRetainedTabletBytes() + queueB.getRetainedTabletBytes()
- <= memoryManager.getTotalMemorySizeInBytes());
+ assertEquals(0L, memoryManager.getOvercommitSizeInBytes());
+ assertEquals(0L, memoryManager.getOversizedEntryCount());
+ assertEquals(
+ oneTabletBytes, queueA.getRetainedTabletBytes() +
queueB.getRetainedTabletBytes());
readerB.currentSearchIndex = 1L;
assertTrue(pendingEntries(queueB).offer(createRequest(1L)));
@@ -323,6 +338,7 @@ public class ConsensusPrefetchingQueueDataNodeMemoryTest {
assertTrue(queueA.ack("consumerA", eventA.getCommitContext()));
assertEquals(0L, queueA.getRetainedTabletBytes());
assertEquals(oneTabletBytes, memoryManager.getUsedMemorySizeInBytes());
+ assertEquals(0L, memoryManager.getOvercommitSizeInBytes());
queueB.drivePrefetchOnce();
assertEquals("false",
queueB.coreReportMessage().get("realtimeAdmissionBlocked"));
@@ -331,6 +347,7 @@ public class ConsensusPrefetchingQueueDataNodeMemoryTest {
assertNotNull(eventB);
assertTrue(queueB.ack("consumerB", eventB.getCommitContext()));
assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
+ assertEquals(0L, memoryManager.getOvercommitSizeInBytes());
queueB.drivePrefetchOnce();
assertEquals("false",
queueB.coreReportMessage().get("realtimeAdmissionBlocked"));
} finally {
@@ -347,6 +364,151 @@ public class ConsensusPrefetchingQueueDataNodeMemoryTest {
}
}
+ @Test
+ public void testWatermarkAndAckDoNotHideDataDeliveryGap() throws Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final int originalBatchMaxDelay =
+
CommonDescriptor.getInstance().getConfig().getSubscriptionConsensusBatchMaxDelayInMs();
+ final File systemDir = temporaryFolder.newFolder("delivery-gap");
+ ConsensusPrefetchingQueue queue = null;
+ try {
+
CommonDescriptor.getInstance().getConfig().setSubscriptionConsensusBatchMaxDelayInMs(0);
+ final FakeConsensusReqReader reader = new FakeConsensusReqReader();
+ queue =
+ newQueue(
+ "consumerGroup",
+ new DataRegionId(1),
+ reader,
+ newConverter(new AtomicInteger()),
+ newCommitManager(systemDir),
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE);
+ final AtomicLong clockNs = new AtomicLong(-1_000_000L);
+ setTimeTracker(queue, new SubscriptionQueueTimeTracker(clockNs::get));
+ queue.setSubscriptionMemoryManager(new
SubscriptionMemoryManager(1_000_000L));
+
+ assertNull(queue.poll("consumer"));
+ reader.currentSearchIndex = 1L;
+ assertTrue(pendingEntries(queue).offer(createRequest(1L)));
+ queue.drivePrefetchOnce();
+ final SubscriptionEvent first = queue.poll("consumer");
+ assertNotNull(first);
+
+ clockNs.addAndGet(TimeUnit.MILLISECONDS.toNanos(60_000L));
+ final Method injectWatermark =
+ ConsensusPrefetchingQueue.class.getDeclaredMethod("injectWatermark",
long.class);
+ injectWatermark.setAccessible(true);
+ injectWatermark.invoke(queue, 1L);
+ final SubscriptionEvent watermark = queue.poll("consumer");
+ assertNotNull(watermark);
+ assertEquals(
+ SubscriptionPollResponseType.WATERMARK.getType(),
+ watermark.getCurrentResponse().getResponseType());
+ assertEquals(60_000L, queue.getDeliveryIdleTimeMs());
+ assertTrue(queue.ack("consumer", first.getCommitContext()));
+ assertEquals(60_000L, queue.getDeliveryIdleTimeMs());
+ assertEquals(0L, queue.getLastDeliveryIntervalMs());
+
+ clockNs.addAndGet(TimeUnit.MILLISECONDS.toNanos(1L));
+ reader.currentSearchIndex = 2L;
+ assertTrue(pendingEntries(queue).offer(createRequest(2L)));
+ queue.drivePrefetchOnce();
+ final SubscriptionEvent second = queue.poll("consumer");
+ assertNotNull(second);
+ assertEquals(60_001L, queue.getLastDeliveryIntervalMs());
+ assertEquals(60_001L, queue.getMaxDeliveryIntervalMs());
+ assertEquals(0L, queue.getDeliveryIdleTimeMs());
+ assertTrue(queue.ack("consumer", second.getCommitContext()));
+
+ clockNs.addAndGet(TimeUnit.MILLISECONDS.toNanos(1L));
+ reader.currentSearchIndex = 3L;
+ assertTrue(pendingEntries(queue).offer(createRequest(3L)));
+ queue.drivePrefetchOnce();
+ assertNotNull(queue.poll("consumer"));
+ assertEquals(1L, queue.getLastDeliveryIntervalMs());
+ assertEquals(60_001L, queue.getMaxDeliveryIntervalMs());
+ } finally {
+ if (queue != null) {
+ queue.close();
+ }
+ CommonDescriptor.getInstance()
+ .getConfig()
+ .setSubscriptionConsensusBatchMaxDelayInMs(originalBatchMaxDelay);
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
+ @Test
+ public void testTimingRemainsReadableDuringBlockedPrefetch() throws
Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final File systemDir =
temporaryFolder.newFolder("blocked-prefetch-timing");
+ final CountDownLatch conversionStarted = new CountDownLatch(1);
+ final CountDownLatch releaseConversion = new CountDownLatch(1);
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ ConsensusPrefetchingQueue queue = null;
+ try {
+ final FakeConsensusReqReader reader = new FakeConsensusReqReader();
+ final ConsensusLogToTabletConverter converter =
mock(ConsensusLogToTabletConverter.class);
+ when(converter.getDatabaseName()).thenReturn("db");
+ when(converter.convert(any()))
+ .thenAnswer(
+ ignored -> {
+ conversionStarted.countDown();
+ assertTrue(releaseConversion.await(5L, TimeUnit.SECONDS));
+ return Collections.singletonList(createTablet());
+ });
+ queue =
+ newQueue(
+ "consumerGroup",
+ new DataRegionId(1),
+ reader,
+ converter,
+ newCommitManager(systemDir),
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE);
+ final AtomicLong clockNs = new AtomicLong(Long.MAX_VALUE - 1_000_000L);
+ setTimeTracker(queue, new SubscriptionQueueTimeTracker(clockNs::get));
+ queue.setSubscriptionMemoryManager(new
SubscriptionMemoryManager(1_000_000L));
+
+ assertNull(queue.poll("consumer"));
+ reader.currentSearchIndex = 1L;
+ assertTrue(pendingEntries(queue).offer(createRequest(1L)));
+ final ConsensusPrefetchingQueue blockedQueue = queue;
+ final Future<PrefetchRoundResult> round =
executor.submit(blockedQueue::drivePrefetchOnce);
+ assertTrue(conversionStarted.await(5L, TimeUnit.SECONDS));
+ clockNs.addAndGet(TimeUnit.MILLISECONDS.toNanos(65_000L));
+
+ assertEquals(65_000L, queue.getPrefetchDurationMs());
+ assertEquals(65_000L, queue.getMaxPrefetchDurationMs());
+ assertEquals(65_000L, queue.getPrefetchIdleTimeMs());
+ assertEquals(65_000L, queue.getDeliveryIdleTimeMs());
+ assertEquals("65000",
queue.coreReportMessage().get("prefetchDurationMs"));
+ assertEquals("0",
queue.coreReportMessage().get("subscriptionMemoryUsedInBytes"));
+ assertEquals("0",
queue.coreReportMessage().get("dataNodeSubscriptionMemoryUsedInBytes"));
+
+ releaseConversion.countDown();
+ round.get(5L, TimeUnit.SECONDS);
+ assertEquals(0L, queue.getPrefetchDurationMs());
+ assertEquals(65_000L, queue.getMaxPrefetchDurationMs());
+ assertEquals(0L, queue.getPrefetchIdleTimeMs());
+ queue.drivePrefetchOnce();
+ assertEquals(65_000L, queue.getMaxPrefetchDurationMs());
+ } finally {
+ releaseConversion.countDown();
+ executor.shutdownNow();
+ if (queue != null) {
+ queue.close();
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
+ private static void setTimeTracker(
+ final ConsensusPrefetchingQueue queue, final
SubscriptionQueueTimeTracker tracker)
+ throws Exception {
+ final Field field =
ConsensusPrefetchingQueue.class.getDeclaredField("timeTracker");
+ field.setAccessible(true);
+ field.set(queue, tracker);
+ }
+
@Test
public void
testRealtimeBacklogIsDrainedBeforeMorePendingEntriesAreConverted() throws
Exception {
final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
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 600073d7e3d..0b9e12eda48 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
@@ -94,7 +94,7 @@ import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
@@ -2388,6 +2388,57 @@ public class ConsensusPrefetchingQueueTest {
}
}
+ @Test
+ public void testPeriodicStatsDoNotReadWal() throws Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final File systemDir = temporaryFolder.newFolder("stats-without-wal-read");
+ ConsensusPrefetchingQueue queue = null;
+ try {
+ final IoTConsensusServerImpl serverImpl =
mock(IoTConsensusServerImpl.class);
+ when(serverImpl.getConsensusReqReader()).thenReturn(new
FakeConsensusReqReader());
+ when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new
WriterSafeFrontierTracker());
+ queue =
+ new ConsensusPrefetchingQueue(
+ "consumerGroup",
+ "topic",
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
+ new DataRegionId(1),
+ serverImpl,
+ new SubscriptionWalRetentionPolicy(
+ "topic",
+ SubscriptionWalRetentionPolicy.UNBOUNDED,
+ SubscriptionWalRetentionPolicy.UNBOUNDED),
+ mock(ConsensusLogToTabletConverter.class),
+ newCommitManager(systemDir),
+ new RegionProgress(Collections.emptyMap()),
+ 1L,
+ 1L,
+ true);
+
+ final ProgressWALIterator iterator = mock(ProgressWALIterator.class);
+ setSubscriptionWalIterator(queue, iterator);
+ final Field lastStatsLogTime =
+
ConsensusPrefetchingQueue.class.getDeclaredField("lastStatsLogTimeMs");
+ lastStatsLogTime.setAccessible(true);
+ lastStatsLogTime.setLong(queue, 0L);
+ final Method logStats =
+
ConsensusPrefetchingQueue.class.getDeclaredMethod("logPeriodicStatsIfNecessary");
+ logStats.setAccessible(true);
+ logStats.invoke(queue);
+
+ verify(iterator).hasBufferedNext();
+ verify(iterator, never()).hasNext();
+ verify(iterator, never()).next();
+ verify(iterator, never()).refresh();
+ assertEquals(1L, queue.getCurrentReadSearchIndex());
+ } finally {
+ if (queue != null) {
+ queue.close();
+ }
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
@Test
public void testPendingCursorAdvanceFastForwardsWalIteratorInPlace() throws
Exception {
final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
@@ -2398,6 +2449,8 @@ public class ConsensusPrefetchingQueueTest {
final IoTConsensusServerImpl serverImpl =
mock(IoTConsensusServerImpl.class);
when(serverImpl.getConsensusReqReader()).thenReturn(reader);
when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new
WriterSafeFrontierTracker());
+ final ConsensusLogToTabletConverter converter =
mock(ConsensusLogToTabletConverter.class);
+ when(converter.convert(any())).thenReturn(Collections.emptyList());
queue =
new ConsensusPrefetchingQueue(
"consumerGroup",
@@ -2409,28 +2462,26 @@ public class ConsensusPrefetchingQueueTest {
"topic",
SubscriptionWalRetentionPolicy.UNBOUNDED,
SubscriptionWalRetentionPolicy.UNBOUNDED),
- mock(ConsensusLogToTabletConverter.class),
+ converter,
newCommitManager(systemDir),
new RegionProgress(Collections.emptyMap()),
1L,
1L,
true);
+ for (long searchIndex = 1L; searchIndex <= 7L; searchIndex++) {
+ assertTrue(pendingEntries(queue).offer(createRequest(searchIndex)));
+ }
+ reader.currentSearchIndex = 7L;
+ assertNull(queue.poll("consumer"));
final ProgressWALIterator staleIterator =
mock(ProgressWALIterator.class);
setSubscriptionWalIterator(queue, staleIterator);
-
- final Method advancePendingCursor =
- ConsensusPrefetchingQueue.class.getDeclaredMethod(
- "advanceLocalCursorFromPendingIfPresent",
IndexedConsensusRequest.class, long.class);
- advancePendingCursor.setAccessible(true);
- advancePendingCursor.invoke(
- queue,
- new IndexedConsensusRequest(7L, Collections.emptyList()),
- queue.getCurrentSeekGeneration());
+ queue.drivePrefetchOnce();
assertEquals(8L, queue.getCurrentReadSearchIndex());
+ verify(converter, times(7)).convert(any());
verify(staleIterator)
- .advanceTo(anyLong(),
any(ProgressWALIterator.WriterProgressCoverage.class));
+ .advanceTo(eq(8L),
any(ProgressWALIterator.WriterProgressCoverage.class));
verify(staleIterator, never()).close();
assertSame(staleIterator, subscriptionWalIterator(queue));
} finally {
@@ -2545,10 +2596,14 @@ public class ConsensusPrefetchingQueueTest {
reader.currentSearchIndex = 3L;
assertNull(queue.poll("consumer"));
+ final ProgressWALIterator iterator = mock(ProgressWALIterator.class);
+ setSubscriptionWalIterator(queue, iterator);
queue.drivePrefetchOnce();
assertEquals(2, conversionCount.get());
assertEquals(2L, queue.getCurrentReadSearchIndex());
+ verify(iterator).advanceTo(eq(2L),
any(ProgressWALIterator.WriterProgressCoverage.class));
+ verify(iterator, never()).close();
assertEquals(oneTabletBytes, queue.getRetainedTabletBytes());
assertTrue(memoryManager.getFreeMemorySizeInBytes() > 0L);
assertEquals(1, queue.getPrefetchedEventCount());
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWalBackpressureTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWalBackpressureTest.java
index e25c77803cf..54ea59fbb0a 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWalBackpressureTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusPrefetchingQueueWalBackpressureTest.java
@@ -66,6 +66,7 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
@@ -210,6 +211,127 @@ public class ConsensusPrefetchingQueueWalBackpressureTest
{
}
}
+ @Test
+ public void testWalMemoryBlockRetriesWithoutRecreatingIterator() throws
Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final CommonConfig config = CommonDescriptor.getInstance().getConfig();
+ final int originalBatchMaxWalEntries =
config.getSubscriptionConsensusBatchMaxWalEntries();
+ final int originalBatchMaxTabletCount =
config.getSubscriptionConsensusBatchMaxTabletCount();
+ final long originalBatchMaxSize =
config.getSubscriptionConsensusBatchMaxSizeInBytes();
+ final int originalBatchMaxDelay =
config.getSubscriptionConsensusBatchMaxDelayInMs();
+ final File systemDir =
temporaryFolder.newFolder("system-wal-memory-retry");
+ final File walDirectory = temporaryFolder.newFolder("wal-memory-retry");
+ ConsensusPrefetchingQueue queue = null;
+ try {
+ final Tablet sampleTablet = createTablet();
+ final long oneTabletBytes = sampleTablet.ramBytesUsed();
+ config.setSubscriptionConsensusBatchMaxWalEntries(128);
+ config.setSubscriptionConsensusBatchMaxTabletCount(128);
+ config.setSubscriptionConsensusBatchMaxSizeInBytes(oneTabletBytes * 16L);
+ config.setSubscriptionConsensusBatchMaxDelayInMs(0);
+
+ writeSealedWal(walDirectory);
+
+ final WALNode walNode = mock(WALNode.class);
+ when(walNode.getLogDirectory()).thenReturn(walDirectory);
+ when(walNode.getCurrentSearchIndex()).thenReturn((long) REQUEST_COUNT);
+ when(walNode.getCurrentWALFileVersion()).thenReturn(1L);
+ when(walNode.getCurrentWALMetaDataSnapshot()).thenReturn(new
WALMetaData());
+
+ final IoTConsensusServerImpl serverImpl =
mock(IoTConsensusServerImpl.class);
+ when(serverImpl.getConsensusReqReader()).thenReturn(walNode);
+ when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new
WriterSafeFrontierTracker());
+
+ final List<Long> conversionAttempts = new ArrayList<>();
+ final ConsensusLogToTabletConverter converter =
mock(ConsensusLogToTabletConverter.class);
+ when(converter.convert(any()))
+ .thenAnswer(
+ invocation -> {
+ final InsertNode insertNode = (InsertNode)
invocation.getArgument(0);
+ conversionAttempts.add(insertNode.getSearchIndex());
+ return Collections.singletonList(createTablet());
+ });
+ when(converter.getDatabaseName()).thenReturn("db");
+
+ final DataRegionId regionId = new DataRegionId(1);
+ final ConsensusSubscriptionCommitManager commitManager =
newCommitManager(systemDir);
+ queue =
+ new ConsensusPrefetchingQueue(
+ "consumerGroup",
+ "topic",
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE,
+ regionId,
+ serverImpl,
+ new SubscriptionWalRetentionPolicy(
+ "topic",
+ SubscriptionWalRetentionPolicy.UNBOUNDED,
+ SubscriptionWalRetentionPolicy.UNBOUNDED),
+ converter,
+ commitManager,
+ new RegionProgress(Collections.emptyMap()),
+ 1L,
+ 1L,
+ true);
+ final TrackingSubscriptionMemoryManager memoryManager =
+ new TrackingSubscriptionMemoryManager(oneTabletBytes + Math.max(1L,
oneTabletBytes / 2L));
+ queue.setSubscriptionMemoryManager(memoryManager);
+
+ assertNull(queue.poll("consumer"));
+ final ProgressWALIterator iterator = subscriptionWalIterator(queue);
+
+ queue.drivePrefetchOnce();
+ assertEquals(Arrays.asList(1L, 2L), conversionAttempts);
+ assertEquals(2L, queue.getCurrentReadSearchIndex());
+ assertTrue(iterator.hasBufferedNext());
+ assertSame(iterator, subscriptionWalIterator(queue));
+ assertMemoryBounded(queue, memoryManager);
+
+ // No memory has been released yet: another round must neither convert
nor lose the request.
+ queue.drivePrefetchOnce();
+ assertEquals(Arrays.asList(1L, 2L), conversionAttempts);
+ assertEquals(2L, queue.getCurrentReadSearchIndex());
+ assertSame(iterator, subscriptionWalIterator(queue));
+
+ for (long expectedLocalSeq = 1L; expectedLocalSeq <= REQUEST_COUNT;
expectedLocalSeq++) {
+ final SubscriptionEvent event = queue.poll("consumer");
+ assertNotNull(event);
+ assertEquals(expectedLocalSeq,
event.getCommitContext().getWriterProgress().getLocalSeq());
+ assertEquals(
+ 1000L + expectedLocalSeq,
+ event.getCommitContext().getWriterProgress().getPhysicalTime());
+ assertTrue(queue.ack("consumer", event.getCommitContext()));
+ assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
+
+ queue.drivePrefetchOnce();
+ assertSame(iterator, subscriptionWalIterator(queue));
+ assertMemoryBounded(queue, memoryManager);
+ }
+
+ assertEquals(Arrays.asList(1L, 2L, 2L, 3L, 3L), conversionAttempts);
+ assertEquals(3L, queue.getWalPathAcceptedEntries());
+ assertEquals(4L, queue.getCurrentReadSearchIndex());
+ assertEquals(0, queue.getPrefetchedEventCount());
+ assertEquals(0L, queue.getSubscriptionUncommittedEventCount());
+ assertEquals(0L, queue.getRetainedTabletBytes());
+ assertNull(queue.poll("consumer"));
+ assertEquals(
+ new WriterProgress(1000L + REQUEST_COUNT, REQUEST_COUNT),
+ commitManager
+ .getCommittedRegionProgress("consumerGroup", "topic", regionId)
+ .getWriterPositions()
+ .get(new WriterId(regionId.toString(), WRITER_NODE_ID)));
+ } finally {
+ if (queue != null) {
+ queue.close();
+ }
+
config.setSubscriptionConsensusBatchMaxWalEntries(originalBatchMaxWalEntries);
+
config.setSubscriptionConsensusBatchMaxTabletCount(originalBatchMaxTabletCount);
+ config.setSubscriptionConsensusBatchMaxSizeInBytes(originalBatchMaxSize);
+ config.setSubscriptionConsensusBatchMaxDelayInMs(originalBatchMaxDelay);
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
private static void writeSealedWal(final File walDirectory) throws Exception
{
final File historicalWal =
new File(
@@ -288,6 +410,13 @@ public class ConsensusPrefetchingQueueWalBackpressureTest {
return (BlockingQueue<IndexedConsensusRequest>) field.get(queue);
}
+ private static ProgressWALIterator subscriptionWalIterator(final
ConsensusPrefetchingQueue queue)
+ throws Exception {
+ final Field field =
ConsensusPrefetchingQueue.class.getDeclaredField("subscriptionWALIterator");
+ field.setAccessible(true);
+ return (ProgressWALIterator) field.get(queue);
+ }
+
private static ConsensusSubscriptionCommitManager newCommitManager(final
File systemDir)
throws Exception {
IoTDBDescriptor.getInstance().getConfig().setSystemDir(systemDir.getAbsolutePath());
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java
index 8af97b31964..d6159e6c692 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/ProgressWALIteratorTest.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.subscription.broker.consensus;
import
org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeType;
import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.SearchNode;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALEntryType;
import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALInfoEntry;
@@ -397,7 +398,8 @@ public class ProgressWALIteratorTest {
final String detail = iterator.getSkippedBrokenWalFileDetails(0);
assertTrue(detail.contains(firstBrokenWal.getName()));
assertTrue(detail.contains(secondBrokenWal.getName()));
- assertTrue(detail.contains("skippedEntries=unknown"));
+ assertTrue(
+ detail.contains("skippedEntries=" +
DataNodePipeMessages.MESSAGE_UNKNOWN_AD921D60));
assertEquals(2L, iterator.next().getSearchIndex());
assertFalse(iterator.hasNext());
assertTrue(iterator.hasSkippedBrokenWalFiles());
@@ -735,6 +737,47 @@ public class ProgressWALIteratorTest {
}
}
+ @Test
+ public void testBufferedNextDoesNotScanPastStaleLocalRequest() throws
Exception {
+ final Path dir =
Files.createTempDirectory("progress-wal-iterator-buffered-next");
+ final File dataWal =
+ dir.resolve(WALFileUtils.getLogFileName(0, 0,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ final File successorWal =
+ dir.resolve(WALFileUtils.getLogFileName(1, 2,
WALFileStatus.CONTAINS_SEARCH_INDEX))
+ .toFile();
+ try {
+ try (WALWriter writer = new WALWriter(dataWal, WALFileVersion.V3)) {
+ writer.write(searchableEntry(1L), singleEntryMeta(19, 1L, 1L, 100L, 7,
1L));
+ writer.write(searchableEntry(-1L), singleEntryMeta(19, -1L, 1L, 200L,
8, 20L));
+ writer.write(searchableEntry(2L), singleEntryMeta(19, 2L, 1L, 300L, 7,
2L));
+ }
+ try (WALWriter ignored = new WALWriter(successorWal, WALFileVersion.V3))
{
+ // Seal the data file so all requests can be read from retained WAL.
+ }
+ try (ProgressWALIterator iterator = new
ProgressWALIterator(dir.toFile(), 1L)) {
+ assertFalse(iterator.hasBufferedNext());
+ assertTrue(iterator.hasNext());
+ assertTrue(iterator.hasBufferedNext());
+
+ iterator.advanceTo(2L, null);
+ assertFalse(iterator.hasBufferedNext());
+ assertTrue(iterator.hasNext());
+ assertTrue(iterator.hasBufferedNext());
+
+ iterator.advanceTo(3L, null);
+ assertTrue(iterator.hasBufferedNext());
+ assertEquals(-1L, iterator.next().getSearchIndex());
+ assertFalse(iterator.hasBufferedNext());
+ assertFalse(iterator.hasNext());
+ }
+ } finally {
+ Files.deleteIfExists(dataWal.toPath());
+ Files.deleteIfExists(successorWal.toPath());
+ Files.deleteIfExists(dir);
+ }
+ }
+
@Test
public void testAdvanceToKeepsUncoveredFollowerRequestInCurrentFile() throws
Exception {
final Path dir =
Files.createTempDirectory("progress-wal-iterator-current-file-advance");
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/SubscriptionWalPrefetchPerformanceTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/SubscriptionWalPrefetchPerformanceTest.java
new file mode 100644
index 00000000000..b268bfd4249
--- /dev/null
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/consensus/SubscriptionWalPrefetchPerformanceTest.java
@@ -0,0 +1,380 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iotdb.db.subscription.broker.consensus;
+
+import org.apache.iotdb.commons.request.IConsensusRequest;
+import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest;
+import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertNode;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode;
+import
org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalInsertRowNode;
+import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils;
+import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALInfoEntry;
+import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALFileVersion;
+import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALMetaData;
+import org.apache.iotdb.db.storageengine.dataregion.wal.io.WALWriter;
+import
org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALByteBufferForTest;
+import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileStatus;
+import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALFileUtils;
+
+import org.junit.Assume;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.lang.reflect.Field;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.Locale;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Manual comparisons of the iterator operations changed by subscription
prefetch optimizations.
+ * Both paths use the current implementation and the same sealed, uncompressed
V3 WAL. Wall time is
+ * diagnostic only; assertions cover work counts, writer progress and retry
payload integrity.
+ */
+public class SubscriptionWalPrefetchPerformanceTest {
+
+ private static final String PREFIX = "iotdb.test.subscription.performance.";
+ private static final int FRAGMENTS = 2;
+
+ @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ /**
+ * Enable with -Diotdb.test.subscription.performance.prefetch-overhead=true.
Models a lagging
+ * iterator and a pending batch that advances past a retained file with an
uncovered follower.
+ */
+ @Test
+ public void testNoIoStatsProbeAndCoalescedAdvancePerformance() throws
Exception {
+ assumeEnabled("prefetch-overhead");
+ final int entryCount = positiveProperty("entries", 4096);
+ final int batchSize = positiveProperty("batch-size", 64);
+ final int iterations = positiveProperty("iterations", 5);
+ final int warmups = positiveProperty("warmups", 2);
+ assertTrue(entryCount >= 2);
+ final File walDirectory = createWal(entryCount, true, batchSize);
+ final long probeIndex = Math.max(1L, entryCount * 3L / 4L);
+
+ final Sample[] probing = new Sample[iterations];
+ final Sample[] peeking = new Sample[iterations];
+ final Sample[] perEntry = new Sample[iterations];
+ final Sample[] coalesced = new Sample[iterations];
+ for (int round = 0; round < warmups + iterations; round++) {
+ // Alternate order so the optimized path does not always follow the
baseline.
+ final boolean optimizedFirst = (round & 1) == 1;
+ final Sample firstProbe = measureStatsProbe(walDirectory, probeIndex,
optimizedFirst);
+ final Sample secondProbe = measureStatsProbe(walDirectory, probeIndex,
!optimizedFirst);
+ final Sample firstAdvance =
+ measureAdvance(walDirectory, entryCount, batchSize, optimizedFirst);
+ final Sample secondAdvance =
+ measureAdvance(walDirectory, entryCount, batchSize, !optimizedFirst);
+ if (round >= warmups) {
+ final int index = round - warmups;
+ probing[index] = optimizedFirst ? secondProbe : firstProbe;
+ peeking[index] = optimizedFirst ? firstProbe : secondProbe;
+ perEntry[index] = optimizedFirst ? secondAdvance : firstAdvance;
+ coalesced[index] = optimizedFirst ? firstAdvance : secondAdvance;
+ }
+ }
+
+ printWorkload(entryCount, warmups, iterations);
+ System.out.printf(Locale.ROOT, "batchSize=%d, probeSearchIndex=%d%n",
batchSize, probeIndex);
+ printComparison("stats probe (work=entry bodies read)", probing, peeking);
+ printComparison("cursor advance (work=footer tuples checked)", perEntry,
coalesced);
+ }
+
+ /**
+ * Enable with -Diotdb.test.subscription.performance.memory-retry=true.
Replays the queue's old
+ * reset and new cache-restoration operations after each rejected
materialization attempt.
+ * Includes deserialization; excludes Tablet conversion, allocation and
scheduler wait time.
+ */
+ @Test
+ public void testMemoryBlockedRetryPerformance() throws Exception {
+ assumeEnabled("memory-retry");
+ final int entryCount = positiveProperty("entries", 4096);
+ final int retryCount = positiveProperty("retries", 8);
+ final int iterations = positiveProperty("iterations", 5);
+ final int warmups = positiveProperty("warmups", 2);
+ assertTrue(entryCount >= 2);
+ final File walDirectory = createWal(entryCount, false, 0);
+ final long retryIndex = Math.max(1L, entryCount * 3L / 4L);
+
+ final Sample[] recreated = new Sample[iterations];
+ final Sample[] cached = new Sample[iterations];
+ for (int round = 0; round < warmups + iterations; round++) {
+ final boolean optimizedFirst = (round & 1) == 1;
+ final Sample first = measureRetries(walDirectory, retryIndex,
retryCount, optimizedFirst);
+ final Sample second = measureRetries(walDirectory, retryIndex,
retryCount, !optimizedFirst);
+ if (round >= warmups) {
+ final int index = round - warmups;
+ recreated[index] = optimizedFirst ? second : first;
+ cached[index] = optimizedFirst ? first : second;
+ }
+ }
+
+ printWorkload(entryCount, warmups, iterations);
+ System.out.printf(Locale.ROOT, "attempts=%d, retrySearchIndex=%d%n",
retryCount, retryIndex);
+ printComparison("memory retry (work=entry bodies read)", recreated,
cached);
+ }
+
+ private File createWal(final int entryCount, final boolean includeFollower,
final int batchSize)
+ throws Exception {
+ final File directory = temporaryFolder.newFolder();
+ final File historical =
+ new File(
+ directory, WALFileUtils.getLogFileName(0L, 0L,
WALFileStatus.CONTAINS_SEARCH_INDEX));
+ try (WALWriter writer = new WALWriter(historical, WALFileVersion.V3)) {
+ for (long index = 1L; index <= entryCount; index++) {
+ writeRequest(writer, index, 7, index);
+ }
+ if (includeFollower) {
+ writeRequest(writer, -1L, 8, 1L);
+ }
+ }
+ final File successor =
+ new File(
+ directory,
+ WALFileUtils.getLogFileName(1L, entryCount,
WALFileStatus.CONTAINS_SEARCH_INDEX));
+ try (WALWriter writer = new WALWriter(successor, WALFileVersion.V3)) {
+ if (includeFollower) {
+ for (long index = entryCount + 1L; index <= entryCount + batchSize +
1L; index++) {
+ writeRequest(writer, index, 7, index);
+ }
+ }
+ // Even an empty successor makes the first file historical and readable.
+ }
+ return directory;
+ }
+
+ private static void writeRequest(
+ final WALWriter writer, final long searchIndex, final int nodeId, final
long localSeq)
+ throws Exception {
+ for (int fragment = 0; fragment < FRAGMENTS; fragment++) {
+ final RelationalInsertRowNode node =
+ StatementTestUtils.genInsertRowNode(Math.toIntExact(localSeq *
FRAGMENTS + fragment));
+ node.setSearchIndex(searchIndex);
+ node.setLastFragment(fragment == FRAGMENTS - 1);
+ final WALInfoEntry entry = new WALInfoEntry(1L, node);
+ final ByteBuffer buffer = ByteBuffer.allocate(entry.serializedSize());
+ entry.serialize(new WALByteBufferForTest(buffer));
+ final WALMetaData metadata = new WALMetaData();
+ metadata.add(entry.serializedSize(), searchIndex, 1L, 1000L + localSeq,
nodeId, localSeq);
+ // Small individual segments intentionally avoid changing global WAL
compression settings.
+ writer.write(buffer, metadata);
+ }
+ }
+
+ private static Sample measureStatsProbe(
+ final File directory, final long probeIndex, final boolean optimized)
throws Exception {
+ final Field bodyCount = bodyCountField();
+ try (ProgressWALIterator iterator = new ProgressWALIterator(directory,
probeIndex)) {
+ final long startNanos = System.nanoTime();
+ final boolean ready = optimized ? iterator.hasBufferedNext() :
iterator.hasNext();
+ final long elapsedNanos = System.nanoTime() - startNanos;
+ assertEquals(!optimized, ready);
+ final long readBodies = bodyCount.getInt(iterator);
+ // One lookahead fragment is read to establish the end of a logical
request.
+ assertEquals(optimized ? 0L : probeIndex * FRAGMENTS + 1L, readBodies);
+ assertRequest(iterator.next(), probeIndex, 7, probeIndex);
+ return new Sample(elapsedNanos, readBodies);
+ }
+ }
+
+ private static Sample measureAdvance(
+ final File directory, final int entryCount, final int batchSize, final
boolean optimized)
+ throws Exception {
+ final long[] footerTuples = {0L};
+ final ProgressWALIterator.WriterProgressCoverage coverage =
+ (physicalTime, nodeId, localSeq) -> {
+ footerTuples[0]++;
+ return nodeId == 7 && localSeq <= entryCount;
+ };
+ try (ProgressWALIterator iterator = new ProgressWALIterator(directory,
1L)) {
+ final long startNanos = System.nanoTime();
+ if (optimized) {
+ iterator.advanceTo(entryCount + batchSize + 1L, coverage);
+ } else {
+ for (int index = 1; index <= batchSize; index++) {
+ iterator.advanceTo(entryCount + index + 1L, coverage);
+ }
+ }
+ final long elapsedNanos = System.nanoTime() - startNanos;
+ // The first follower tuple prevents skipping the retained file on every
coverage check.
+ assertEquals(
+ (entryCount * (long) FRAGMENTS + 1L) * (optimized ? 1L : batchSize),
footerTuples[0]);
+ assertRequest(iterator.next(), -1L, 8, 1L);
+ final long nextIndex = entryCount + batchSize + 1L;
+ assertRequest(iterator.next(), nextIndex, 7, nextIndex);
+ assertFalse(iterator.hasNext());
+ return new Sample(elapsedNanos, footerTuples[0]);
+ }
+ }
+
+ private static Sample measureRetries(
+ final File directory, final long retryIndex, final int attempts, final
boolean optimized)
+ throws Exception {
+ final Field bodyCount = bodyCountField();
+ ProgressWALIterator iterator = null;
+ IndexedConsensusRequest previousRequest = null;
+ long readBodies = 0L;
+ int previousBodyCount = 0;
+ try {
+ final long startNanos = System.nanoTime();
+ for (int attempt = 0; attempt < attempts; attempt++) {
+ if (iterator == null) {
+ iterator = new ProgressWALIterator(directory, retryIndex);
+ }
+ final IndexedConsensusRequest request = iterator.next();
+ assertRequest(request, retryIndex, 7, retryIndex);
+ final int currentBodyCount = bodyCount.getInt(iterator);
+ readBodies += optimized ? currentBodyCount - previousBodyCount :
currentBodyCount;
+ previousBodyCount = currentBodyCount;
+ if (optimized && previousRequest != null) {
+ assertSame(previousRequest, request);
+ }
+ previousRequest = request;
+ if (attempt + 1 < attempts) {
+ if (optimized) {
+ iterator.retry(request);
+ } else {
+ iterator.close();
+ iterator = null;
+ }
+ }
+ }
+ final long elapsedNanos = System.nanoTime() - startNanos;
+ assertEquals((retryIndex * FRAGMENTS + 1L) * (optimized ? 1L :
attempts), readBodies);
+ // The last attempt succeeds: the pending lookahead must still lead to
the next request.
+ assertRequest(iterator.next(), retryIndex + 1L, 7, retryIndex + 1L);
+ return new Sample(elapsedNanos, readBodies);
+ } finally {
+ if (iterator != null) {
+ iterator.close();
+ }
+ }
+ }
+
+ private static void assertRequest(
+ final IndexedConsensusRequest request,
+ final long searchIndex,
+ final int nodeId,
+ final long localSeq) {
+ assertEquals(searchIndex, request.getSearchIndex());
+ assertEquals(nodeId, request.getNodeId());
+ assertEquals(localSeq, request.getProgressLocalSeq());
+ assertEquals(1000L + localSeq, request.getPhysicalTime());
+ assertEquals(FRAGMENTS, request.getRequests().size());
+ final InsertNode insert =
ConsensusLogToTabletConverter.deserializeToInsertNode(request);
+ assertNotNull(insert);
+ assertEquals(searchIndex, insert.getSearchIndex());
+ assertEquals(nodeId, insert.getNodeId());
+ assertEquals(localSeq, insert.getSyncIndex());
+ assertEquals(1000L + localSeq, insert.getPhysicalTime());
+ assertTrue(insert instanceof InsertRowsNode);
+ final InsertRowsNode rows = (InsertRowsNode) insert;
+ assertEquals(FRAGMENTS, rows.getInsertRowNodeList().size());
+ for (int fragment = 0; fragment < FRAGMENTS; fragment++) {
+ final InsertRowNode row = rows.getInsertRowNodeList().get(fragment);
+ final long expectedValue = localSeq * FRAGMENTS + fragment;
+ assertEquals(expectedValue, row.getTime());
+ assertEquals((double) expectedValue, (Double) row.getValues()[2], 0.0);
+ }
+ for (final IConsensusRequest fragment : request.getRequests()) {
+ assertEquals(0, fragment.serializeToByteBuffer().position());
+ }
+ }
+
+ private static Field bodyCountField() throws Exception {
+ // All counted reads remain in one historical file; inspect existing state
without a
+ // production instrumentation hook or overhead in WALReader.next().
+ final Field field =
+
ProgressWALIterator.class.getDeclaredField("consumedEntryCountInCurrentFile");
+ field.setAccessible(true);
+ return field;
+ }
+
+ private static void assumeEnabled(final String benchmark) {
+ Assume.assumeTrue(
+ "Enable with -D" + PREFIX + benchmark + "=true",
Boolean.getBoolean(PREFIX + benchmark));
+ }
+
+ private static int positiveProperty(final String name, final int
defaultValue) {
+ final int value = Integer.getInteger(PREFIX + name, defaultValue);
+ assertTrue(name + " must be positive", value > 0);
+ return value;
+ }
+
+ private static void printWorkload(final int entryCount, final int warmups,
final int iterations) {
+ System.out.printf(
+ Locale.ROOT,
+ "Subscription WAL benchmark: requests=%d, fragments=%d, warmups=%d,
samples=%d, "
+ + "os=%s, java=%s%n",
+ entryCount,
+ FRAGMENTS,
+ warmups,
+ iterations,
+ System.getProperty("os.name"),
+ System.getProperty("java.version"));
+ }
+
+ private static void printComparison(
+ final String name, final Sample[] baseline, final Sample[] optimized) {
+ final double baselineNanos = medianNanos(baseline);
+ final double optimizedNanos = medianNanos(optimized);
+ System.out.printf(
+ Locale.ROOT,
+ "%s: baselineMedian=%.6f ms, optimizedMedian=%.6f ms, speedup=%.2fx, "
+ + "baselineWork=%d, optimizedWork=%d%n",
+ name,
+ baselineNanos / 1_000_000.0,
+ optimizedNanos / 1_000_000.0,
+ baselineNanos / Math.max(1.0, optimizedNanos),
+ baseline[0].work,
+ optimized[0].work);
+ }
+
+ private static double medianNanos(final Sample[] samples) {
+ final long[] values = new long[samples.length];
+ for (int index = 0; index < samples.length; index++) {
+ values[index] = samples[index].elapsedNanos;
+ }
+ Arrays.sort(values);
+ final int middle = values.length / 2;
+ return values.length % 2 == 0
+ ? values[middle - 1] + (values[middle] - values[middle - 1]) / 2.0
+ : values[middle];
+ }
+
+ private static final class Sample {
+ private final long elapsedNanos;
+ private final long work;
+
+ private Sample(final long elapsedNanos, final long work) {
+ this.elapsedNanos = elapsedNanos;
+ this.work = work;
+ }
+ }
+}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/SubscriptionMetricsRestartTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/SubscriptionMetricsRestartTest.java
index a2142ad61fd..68eee16ee2f 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/SubscriptionMetricsRestartTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/SubscriptionMetricsRestartTest.java
@@ -24,6 +24,8 @@ import org.apache.iotdb.commons.service.metric.MetricService;
import org.apache.iotdb.commons.service.metric.enums.Metric;
import org.apache.iotdb.db.subscription.broker.SubscriptionPrefetchingQueue;
import
org.apache.iotdb.db.subscription.broker.consensus.ConsensusPrefetchingQueue;
+import
org.apache.iotdb.db.subscription.resource.SubscriptionDataNodeResourceManager;
+import org.apache.iotdb.db.subscription.resource.SubscriptionMemoryManager;
import org.apache.iotdb.metrics.config.MetricConfig;
import org.apache.iotdb.metrics.config.MetricConfigDescriptor;
import org.apache.iotdb.metrics.metricsets.IMetricSet;
@@ -38,12 +40,15 @@ import org.junit.Before;
import org.junit.Test;
import org.mockito.Mockito;
+import java.lang.reflect.Field;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
import static org.mockito.Mockito.when;
/** Subscription metrics must survive a metric service restart, which drops
all metrics. */
@@ -96,6 +101,52 @@ public class SubscriptionMetricsRestartTest {
assertEquals(0, count(Metric.SUBSCRIPTION_EVENT_TRANSFER));
}
+ @Test
+ public void testSubscriptionMemoryMetrics() throws Exception {
+ final SubscriptionMetrics metrics = SubscriptionMetrics.getInstance();
+ // Subscription can be disabled in the default test configuration. Give
the shared allocator
+ // an explicit test budget and restore its original block after removing
the metrics.
+ final long budget = 10L;
+ final SubscriptionMemoryManager memoryManager =
SubscriptionDataNodeResourceManager.memory();
+ final long originalOversizedEntryCount =
memoryManager.getOversizedEntryCount();
+ final Field blockField =
SubscriptionMemoryManager.class.getDeclaredField("memoryBlock");
+ blockField.setAccessible(true);
+ final Object originalBlock = blockField.get(memoryManager);
+ try {
+ blockField.set(memoryManager, blockField.get(new
SubscriptionMemoryManager(budget)));
+ bind(metrics);
+ assertTrue(memoryManager.tryAllocate(budget + 1L));
+ try {
+ service.restartService();
+ assertEquals(
+ budget, ((AutoGauge)
get(Metric.SUBSCRIPTION_MEMORY_LIMIT_BYTES)).getValue(), 0);
+ assertEquals(
+ budget + 1L, ((AutoGauge)
get(Metric.SUBSCRIPTION_MEMORY_USED_BYTES)).getValue(), 0);
+ assertEquals(
+ 1, ((AutoGauge)
get(Metric.SUBSCRIPTION_MEMORY_OVERCOMMIT_BYTES)).getValue(), 0);
+ assertEquals(
+ originalOversizedEntryCount + 1L,
+ ((AutoGauge)
get(Metric.SUBSCRIPTION_MEMORY_OVERSIZED_ENTRY_COUNT)).getValue(),
+ 0);
+ assertFalse(memoryManager.tryAllocate(1L));
+ } finally {
+ memoryManager.release(budget + 1L);
+ }
+ assertEquals(0, ((AutoGauge)
get(Metric.SUBSCRIPTION_MEMORY_USED_BYTES)).getValue(), 0);
+ assertEquals(0, ((AutoGauge)
get(Metric.SUBSCRIPTION_MEMORY_OVERCOMMIT_BYTES)).getValue(), 0);
+ service.removeMetricSet(metrics);
+ boundMetricSets.remove(metrics);
+ assertEquals(0, count(Metric.SUBSCRIPTION_MEMORY_USED_BYTES));
+ assertEquals(0, count(Metric.SUBSCRIPTION_MEMORY_LIMIT_BYTES));
+ assertEquals(0, count(Metric.SUBSCRIPTION_MEMORY_OVERCOMMIT_BYTES));
+ assertEquals(0, count(Metric.SUBSCRIPTION_MEMORY_OVERSIZED_ENTRY_COUNT));
+ } finally {
+ service.removeMetricSet(metrics);
+ boundMetricSets.remove(metrics);
+ blockField.set(memoryManager, originalBlock);
+ }
+ }
+
@Test
public void testConsensusPrefetchingQueueMetrics() {
final ConsensusSubscriptionPrefetchingQueueMetrics metrics =
@@ -105,16 +156,52 @@ public class SubscriptionMetricsRestartTest {
when(queue.getPrefetchingQueueId()).thenReturn(QUEUE_ID);
when(queue.getConsensusGroupId()).thenReturn(new DataRegionId(1));
when(queue.getLag()).thenReturn(7L);
+ when(queue.getLastDeliveryIntervalMs()).thenReturn(60_001L);
+ when(queue.getMaxDeliveryIntervalMs()).thenReturn(65_000L);
+ when(queue.getDeliveryIdleTimeMs()).thenReturn(60_000L);
+ when(queue.getPrefetchDurationMs()).thenReturn(65_000L);
+ when(queue.getMaxPrefetchDurationMs()).thenReturn(69_000L);
+ when(queue.getPrefetchIdleTimeMs()).thenReturn(67_000L);
metrics.register(queue);
try {
service.restartService();
assertEquals(7, ((AutoGauge)
get(Metric.SUBSCRIPTION_CONSENSUS_LAG)).getValue(), 0);
+ assertEquals(
+ 60_001,
+ ((AutoGauge)
get(Metric.SUBSCRIPTION_CONSENSUS_LAST_DELIVERY_INTERVAL_MS)).getValue(),
+ 0);
+ assertEquals(
+ 65_000,
+ ((AutoGauge)
get(Metric.SUBSCRIPTION_CONSENSUS_MAX_DELIVERY_INTERVAL_MS)).getValue(),
+ 0);
+ assertEquals(
+ 60_000,
+ ((AutoGauge)
get(Metric.SUBSCRIPTION_CONSENSUS_DELIVERY_IDLE_TIME_MS)).getValue(),
+ 0);
+ assertEquals(
+ 65_000,
+ ((AutoGauge)
get(Metric.SUBSCRIPTION_CONSENSUS_PREFETCH_DURATION_MS)).getValue(),
+ 0);
+ assertEquals(
+ 69_000,
+ ((AutoGauge)
get(Metric.SUBSCRIPTION_CONSENSUS_MAX_PREFETCH_DURATION_MS)).getValue(),
+ 0);
+ assertEquals(
+ 67_000,
+ ((AutoGauge)
get(Metric.SUBSCRIPTION_CONSENSUS_PREFETCH_IDLE_TIME_MS)).getValue(),
+ 0);
metrics.mark(QUEUE_ID, new DataRegionId(1).toString(), 5);
assertEquals(5, ((Rate)
get(Metric.SUBSCRIPTION_EVENT_TRANSFER)).getCount());
} finally {
metrics.deregister(queue);
}
assertEquals(0, count(Metric.SUBSCRIPTION_CONSENSUS_LAG));
+ assertEquals(0,
count(Metric.SUBSCRIPTION_CONSENSUS_LAST_DELIVERY_INTERVAL_MS));
+ assertEquals(0,
count(Metric.SUBSCRIPTION_CONSENSUS_MAX_DELIVERY_INTERVAL_MS));
+ assertEquals(0,
count(Metric.SUBSCRIPTION_CONSENSUS_DELIVERY_IDLE_TIME_MS));
+ assertEquals(0, count(Metric.SUBSCRIPTION_CONSENSUS_PREFETCH_DURATION_MS));
+ assertEquals(0,
count(Metric.SUBSCRIPTION_CONSENSUS_MAX_PREFETCH_DURATION_MS));
+ assertEquals(0,
count(Metric.SUBSCRIPTION_CONSENSUS_PREFETCH_IDLE_TIME_MS));
assertEquals(0, count(Metric.SUBSCRIPTION_EVENT_TRANSFER));
}
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/resource/SubscriptionMemoryManagerTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/resource/SubscriptionMemoryManagerTest.java
index beda24e623b..31129d357bb 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/resource/SubscriptionMemoryManagerTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/resource/SubscriptionMemoryManagerTest.java
@@ -35,6 +35,8 @@ public class SubscriptionMemoryManagerTest {
assertFalse(memoryManager.tryAllocate(5L));
assertEquals(6L, memoryManager.getUsedMemorySizeInBytes());
assertEquals(4L, memoryManager.getFreeMemorySizeInBytes());
+ assertEquals(0L, memoryManager.getOvercommitSizeInBytes());
+ assertEquals(0L, memoryManager.getOversizedEntryCount());
memoryManager.release(6L);
assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
@@ -48,10 +50,19 @@ public class SubscriptionMemoryManagerTest {
assertTrue(memoryManager.tryAllocate(11L));
assertFalse(memoryManager.tryAllocate(1L));
assertEquals(11L, memoryManager.getUsedMemorySizeInBytes());
+ assertEquals(1L, memoryManager.getOvercommitSizeInBytes());
+ assertEquals(1L, memoryManager.getOversizedEntryCount());
memoryManager.release(11L);
assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
+ assertEquals(0L, memoryManager.getOvercommitSizeInBytes());
assertTrue(memoryManager.tryAllocate(10L));
+ assertFalse(memoryManager.tryAllocate(11L));
+ assertEquals(1L, memoryManager.getOversizedEntryCount());
+ memoryManager.release(10L);
+ assertTrue(memoryManager.tryAllocate(12L));
+ assertEquals(2L, memoryManager.getOvercommitSizeInBytes());
+ assertEquals(2L, memoryManager.getOversizedEntryCount());
}
@Test
@@ -60,6 +71,7 @@ public class SubscriptionMemoryManagerTest {
assertFalse(memoryManager.tryAllocate(1L));
assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
+ assertEquals(0L, memoryManager.getOversizedEntryCount());
}
@Test
@@ -133,6 +145,8 @@ public class SubscriptionMemoryManagerTest {
SubscriptionMemoryManager.AllocationRejectionReason.OVERSIZED_ENTRY,
queue.tryAllocate(11L).getRejectionReason());
assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
+ assertEquals(0L, memoryManager.getOvercommitSizeInBytes());
+ assertEquals(0L, memoryManager.getOversizedEntryCount());
assertTrue(queue.tryAllocate(10L).isAccepted());
}
assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
index e20919801e4..e6569beb7e6 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Metric.java
@@ -233,7 +233,20 @@ public enum Metric {
SUBSCRIPTION_CONSENSUS_PENDING_EVENT_COUNT("subscription_consensus_pending_event_count"),
SUBSCRIPTION_CONSENSUS_LAST_POLL_TIME("subscription_consensus_last_poll_time"),
SUBSCRIPTION_CONSENSUS_LAST_PROGRESS_TIME("subscription_consensus_last_progress_time"),
+ SUBSCRIPTION_CONSENSUS_LAST_DELIVERY_INTERVAL_MS(
+ "subscription_consensus_last_delivery_interval_ms"),
+ SUBSCRIPTION_CONSENSUS_MAX_DELIVERY_INTERVAL_MS(
+ "subscription_consensus_max_delivery_interval_ms"),
+
SUBSCRIPTION_CONSENSUS_DELIVERY_IDLE_TIME_MS("subscription_consensus_delivery_idle_time_ms"),
+
SUBSCRIPTION_CONSENSUS_PREFETCH_DURATION_MS("subscription_consensus_prefetch_duration_ms"),
+ SUBSCRIPTION_CONSENSUS_MAX_PREFETCH_DURATION_MS(
+ "subscription_consensus_max_prefetch_duration_ms"),
+
SUBSCRIPTION_CONSENSUS_PREFETCH_IDLE_TIME_MS("subscription_consensus_prefetch_idle_time_ms"),
SUBSCRIPTION_CONSENSUS_PROGRESS_STATUS("subscription_consensus_progress_status"),
+ SUBSCRIPTION_MEMORY_USED_BYTES("subscription_memory_used_bytes"),
+ SUBSCRIPTION_MEMORY_LIMIT_BYTES("subscription_memory_limit_bytes"),
+ SUBSCRIPTION_MEMORY_OVERCOMMIT_BYTES("subscription_memory_overcommit_bytes"),
+
SUBSCRIPTION_MEMORY_OVERSIZED_ENTRY_COUNT("subscription_memory_oversized_entry_count"),
SUBSCRIPTION_CONSENSUS_MEMORY_USED("subscription_consensus_memory_used"),
SUBSCRIPTION_CONSENSUS_MEMORY_QUOTA("subscription_consensus_memory_quota"),
SUBSCRIPTION_CONSENSUS_MEMORY_MAXIMUM("subscription_consensus_memory_maximum"),
