This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch fix/subscription-stall-observability in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit e3dc70598bef5cf3ae8106c72d3ed69dcf10d882 Author: Caideyipi <[email protected]> AuthorDate: Fri Oct 9 14:14:07 2026 +0800 Add subscription delivery stall and memory overcommit diagnostics --- .../apache/iotdb/db/i18n/DataNodePipeMessages.java | 8 + .../apache/iotdb/db/i18n/DataNodePipeMessages.java | 8 + .../consensus/ConsensusPrefetchingQueue.java | 64 ++++++++ .../consensus/SubscriptionQueueTimeTracker.java | 115 +++++++++++++ ...nsensusSubscriptionPrefetchingQueueMetrics.java | 60 +++++++ .../subscription/metric/SubscriptionMetrics.java | 34 ++++ .../resource/SubscriptionMemoryManager.java | 37 +++++ ...onsensusPrefetchingQueueDataNodeMemoryTest.java | 180 ++++++++++++++++++++- .../metric/SubscriptionMetricsRestartTest.java | 87 ++++++++++ .../resource/SubscriptionMemoryManagerTest.java | 12 ++ .../iotdb/commons/service/metric/enums/Metric.java | 13 ++ 11 files changed, 611 insertions(+), 7 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 1c0362b627b..e4f487f3fac 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,14 @@ package org.apache.iotdb.db.i18n; public final class DataNodePipeMessages { + 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."; 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 49096659ade..2a94519e41c 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,14 @@ package org.apache.iotdb.db.i18n; public final class DataNodePipeMessages { + 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 失败。"; 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 b692ec94ed0..a2a3871080e 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 @@ -285,6 +285,8 @@ public class ConsensusPrefetchingQueue { private volatile long lastProgressTimeMs = 0L; + private final SubscriptionQueueTimeTracker timeTracker = new SubscriptionQueueTimeTracker(); + private volatile String lastConsumerId = ""; private volatile long lastPendingAcceptedEntries = 0L; @@ -838,6 +840,7 @@ public class ConsensusPrefetchingQueue { this.subscriptionWALIterator = createSubscriptionWALIterator(resolvedStart.getStartSearchIndex()); this.prefetchInitialized = true; + timeTracker.start(); this.observedSeekGeneration = seekGeneration.get(); discardBatch(this.lingerBatch); resetBatchWriterProgress(); @@ -1321,6 +1324,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); @@ -1406,7 +1410,28 @@ 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; + public PrefetchRoundResult drivePrefetchOnce() { + timeTracker.beginPrefetch(); + try { + return drivePrefetchOnceInternal(); + } finally { + final long durationMs = timeTracker.endPrefetch(); + 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(); } @@ -4135,6 +4160,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()) { @@ -4467,6 +4516,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( + "subscriptionMemoryUsedInBytes", + 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/SubscriptionQueueTimeTracker.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/SubscriptionQueueTimeTracker.java new file mode 100644 index 00000000000..350759b1205 --- /dev/null +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/SubscriptionQueueTimeTracker.java @@ -0,0 +1,115 @@ +/* + * 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; + } + } + + synchronized void recordDelivery() { + start(); + final long nowNs = nanoTime.getAsLong(); + if (delivered) { + lastDeliveryIntervalMs = elapsedMs(nowNs, lastDeliveryTimeNs); + maxDeliveryIntervalMs = Math.max(maxDeliveryIntervalMs, lastDeliveryIntervalMs); + } + lastDeliveryTimeNs = nowNs; + delivered = true; + } + + synchronized void beginPrefetch() { + start(); + prefetchStartTimeNs = nanoTime.getAsLong(); + prefetchRunning = true; + } + + synchronized 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 19076e9abae..27325ab602b 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 @@ -193,6 +193,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, @@ -305,6 +341,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 bf71fad6c4b..e33053baa2c 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,13 +24,27 @@ 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.concurrent.TimeUnit; 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 volatile long oversizedEntryCount = 0L; + + private long lastOvercommitWarningTimeNs; + SubscriptionMemoryManager() { memoryBlock = IoTDBDescriptor.getInstance() @@ -52,6 +66,9 @@ public class SubscriptionMemoryManager { * <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 * consensus 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) { if (sizeInBytes <= 0L) { @@ -64,6 +81,18 @@ public class SubscriptionMemoryManager { && memoryBlock.getUsedMemoryInBytes() == 0L && sizeInBytes > memoryBlock.getTotalMemorySizeInBytes()) { memoryBlock.forceAllocateWithoutLimitation(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 true; } return false; @@ -86,4 +115,12 @@ public class SubscriptionMemoryManager { public long getFreeMemorySizeInBytes() { return memoryBlock.getFreeMemoryInBytes(); } + + public long getOvercommitSizeInBytes() { + return Math.max(0L, getUsedMemorySizeInBytes() - getTotalMemorySizeInBytes()); + } + + public long getOversizedEntryCount() { + return oversizedEntryCount; + } } 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 60f30834cad..f9c56e7b56f 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 @@ -33,6 +33,7 @@ import org.apache.iotdb.db.subscription.event.SubscriptionEvent; import org.apache.iotdb.db.subscription.resource.SubscriptionMemoryManager; import org.apache.iotdb.rpc.subscription.config.TopicConstant; 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; @@ -45,12 +46,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; @@ -67,6 +75,15 @@ public class ConsensusPrefetchingQueueDataNodeMemoryTest { @Test public void testQueuesShareDataNodeMemoryBudget() throws Exception { + assertQueuesShareDataNodeMemoryBudget(false); + } + + @Test + public void testOversizedEntryBlocksOtherQueueUntilAck() throws Exception { + assertQueuesShareDataNodeMemoryBudget(true); + } + + private void assertQueuesShareDataNodeMemoryBudget(final boolean oversized) throws Exception { final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); final int originalBatchMaxDelay = CommonDescriptor.getInstance().getConfig().getSubscriptionConsensusBatchMaxDelayInMs(); @@ -77,7 +94,8 @@ public class ConsensusPrefetchingQueueDataNodeMemoryTest { CommonDescriptor.getInstance().getConfig().setSubscriptionConsensusBatchMaxDelayInMs(0); final Tablet tablet = createTablet(); final long oneTabletBytes = PipeMemoryWeightUtil.calculateTabletSizeInBytes(tablet); - final SubscriptionMemoryManager memoryManager = new SubscriptionMemoryManager(oneTabletBytes); + final SubscriptionMemoryManager memoryManager = + new SubscriptionMemoryManager(oneTabletBytes - (oversized ? 1L : 0L)); final ConsensusSubscriptionCommitManager commitManager = newCommitManager(systemDir); final FakeConsensusReqReader readerA = new FakeConsensusReqReader(); @@ -116,9 +134,10 @@ public class ConsensusPrefetchingQueueDataNodeMemoryTest { assertEquals(oneTabletBytes, queueA.getRetainedTabletBytes()); assertEquals(0L, queueB.getRetainedTabletBytes()); assertEquals(oneTabletBytes, memoryManager.getUsedMemorySizeInBytes()); - assertTrue( - queueA.getRetainedTabletBytes() + queueB.getRetainedTabletBytes() - <= memoryManager.getTotalMemorySizeInBytes()); + assertEquals(oversized ? 1L : 0L, memoryManager.getOvercommitSizeInBytes()); + assertEquals(oversized ? 1L : 0L, memoryManager.getOversizedEntryCount()); + assertEquals( + oneTabletBytes, queueA.getRetainedTabletBytes() + queueB.getRetainedTabletBytes()); readerB.currentSearchIndex = 1L; assertFalse(pendingEntries(queueB).offer(createRequest(1L))); @@ -132,6 +151,7 @@ public class ConsensusPrefetchingQueueDataNodeMemoryTest { assertTrue(queueA.ack("consumerA", eventA.getCommitContext())); assertEquals(0L, queueA.getRetainedTabletBytes()); assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + assertEquals(0L, memoryManager.getOvercommitSizeInBytes()); queueB.drivePrefetchOnce(); assertEquals("false", queueB.coreReportMessage().get("realtimeAdmissionBlocked")); @@ -141,14 +161,16 @@ public class ConsensusPrefetchingQueueDataNodeMemoryTest { assertEquals(1, conversionCountB.get()); assertEquals(oneTabletBytes, queueB.getRetainedTabletBytes()); assertEquals(oneTabletBytes, memoryManager.getUsedMemorySizeInBytes()); - assertTrue( - queueA.getRetainedTabletBytes() + queueB.getRetainedTabletBytes() - <= memoryManager.getTotalMemorySizeInBytes()); + assertEquals(oversized ? 1L : 0L, memoryManager.getOvercommitSizeInBytes()); + assertEquals(oversized ? 2L : 0L, memoryManager.getOversizedEntryCount()); + assertEquals( + oneTabletBytes, queueA.getRetainedTabletBytes() + queueB.getRetainedTabletBytes()); final SubscriptionEvent eventB = queueB.poll("consumerB"); assertNotNull(eventB); assertTrue(queueB.ack("consumerB", eventB.getCommitContext())); assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + assertEquals(0L, memoryManager.getOvercommitSizeInBytes()); } finally { if (queueA != null) { queueA.close(); @@ -163,6 +185,150 @@ 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")); + + 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/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 1635a6e0ecf..5b77b5d1406 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,5 +71,6 @@ public class SubscriptionMemoryManagerTest { assertFalse(memoryManager.tryAllocate(1L)); assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + assertEquals(0L, memoryManager.getOversizedEntryCount()); } } 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 5bcafffaa14..5d2a3c69b7b 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 @@ -231,7 +231,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"), // load related ACTIVE_LOADING_FILES_NUMBER("active_loading_files_number"), ACTIVE_LOADING_FILES_SIZE("active_loading_files_size"),
