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 21d53dd015c Bound subscription memory borrowing and report admission
failures (#18812)
21d53dd015c is described below
commit 21d53dd015c43c230914076619cc34ca636a3882
Author: Caideyipi <[email protected]>
AuthorDate: Sat Oct 10 11:17:17 2026 +0800
Bound subscription memory borrowing and report admission failures (#18812)
* Isolate subscription materialization memory by queue
* Allow bounded borrowing of idle subscription memory
---
.../iotdb/consensus/i18n/IoTConsensusMessages.java | 4 +
.../iotdb/consensus/i18n/IoTConsensusMessages.java | 4 +
.../subscription/SubscriptionQueueAdmission.java | 25 ++
.../subscription/SubscriptionQueueRegistry.java | 39 ++-
.../SubscriptionQueueRejectionReason.java | 42 +++
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 2 +
.../apache/iotdb/db/i18n/DataNodePipeMessages.java | 2 +
.../consensus/ConsensusPrefetchingQueue.java | 258 +++++++++++++---
...nsensusSubscriptionPrefetchingQueueMetrics.java | 90 ++++++
.../resource/SubscriptionMemoryManager.java | 335 +++++++++++++++++++--
...onsensusPrefetchingQueueDataNodeMemoryTest.java | 249 +++++++++++++--
.../consensus/ConsensusPrefetchingQueueTest.java | 115 ++++++-
...nsensusPrefetchingQueueWalBackpressureTest.java | 19 +-
...susSubscriptionPrefetchingQueueMetricsTest.java | 23 ++
.../resource/SubscriptionMemoryManagerTest.java | 108 +++++++
.../conf/iotdb-system.properties.template | 29 ++
.../iotdb/commons/service/metric/enums/Metric.java | 8 +
.../iotdb/commons/service/metric/enums/Tag.java | 1 +
18 files changed, 1231 insertions(+), 122 deletions(-)
diff --git
a/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java
b/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java
index 7f6efd07aeb..6c1e4c1dc17 100644
---
a/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java
+++
b/iotdb-core/consensus/src/main/i18n/en/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java
@@ -330,6 +330,10 @@ public final class IoTConsensusMessages {
public static final String
LOG_SUBSCRIPTION_QUEUE_FULL_DROPPED_ARG_ENTRY_S_LAST_ARG_MS_2AD8AB3D =
"Subscription queue full, dropped {} entry(s) in the last {} ms, latest ";
public static final String
LOG_SEARCHINDEX_ARG_QUEUESIZE_ARG_QUEUEREMAINING_ARG_2EA619ED =
"searchIndex={}, queueSize={}, queueRemaining={}";
public static final String
LOG_SUBSCRIPTION_QUEUE_FULL_DROPPED_ENTRY_SEARCHINDEX_ARG_DROPPEDCOUNT_ARG_61F126B8
= "Subscription queue full, dropped entry searchIndex={}, droppedCount={}";
+ public static final String
LOG_SUBSCRIPTION_REALTIME_ADMISSION_REJECTED_ARG_ENTRY_S_IN_THE_LAST_ARG_MS_WAL_REPLAY_REQUIRED_GROUP_ARG_LATEST_SEARCHINDEX_ARG_REASONCODE_ARG_QUEUESIZE_ARG_QUEUEREMAINING_ARG_0FBF6226
=
+ "Subscription realtime admission rejected {} entry(s) in the last {} ms;
WAL replay required, group={}, latest searchIndex={}, reasonCode={},
queueSize={}, queueRemaining={}";
+ public static final String
LOG_SUBSCRIPTION_REALTIME_ADMISSION_REJECTED_ENTRY_WAL_REPLAY_REQUIRED_GROUP_ARG_SEARCHINDEX_ARG_REASONCODE_ARG_REJECTEDCOUNT_ARG_7F76D6A9
=
+ "Subscription realtime admission rejected entry; WAL replay required,
group={}, searchIndex={}, reasonCode={}, rejectedCount={}";
public static final String
LOG_RESERVED_ARG_BYTES_BATCH_ARG_ARG_CURRENT_TOTAL_USAGE_ARG_308AE9C2 =
"Reserved {} bytes for batch {}-{}, current total usage {}";
public static final String
LOG_ARG_FAILED_SEND_IDLE_WRITER_SAFE_TIME_BARRIER_ARG_STATUS_AE047EAD = "{}:
Failed to send idle writer safe-time barrier to {}. status={}";
public static final String
LOG_ARG_WRITE_OPERATION_FAILED_SEARCHINDEX_ARG_CODE_ARG_SUBSCRIPTIONQUEUES_ARG_THIS_ARG_F4B17576
= "{}: write operation failed. searchIndex: {}. Code: {}, subscriptionQueues:
{}, this: {}";
diff --git
a/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java
b/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java
index 7cc8ac622d1..a7ef49bf467 100644
---
a/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java
+++
b/iotdb-core/consensus/src/main/i18n/zh/org/apache/iotdb/consensus/i18n/IoTConsensusMessages.java
@@ -328,6 +328,10 @@ public final class IoTConsensusMessages {
public static final String
LOG_SUBSCRIPTION_QUEUE_FULL_DROPPED_ARG_ENTRY_S_LAST_ARG_MS_2AD8AB3D =
"订阅队列已满，丢弃了 {} 个 entry，最近 {} ms，最新 ";
public static final String
LOG_SEARCHINDEX_ARG_QUEUESIZE_ARG_QUEUEREMAINING_ARG_2EA619ED =
"searchIndex={}, queueSize={}, queueRemaining={}";
public static final String
LOG_SUBSCRIPTION_QUEUE_FULL_DROPPED_ENTRY_SEARCHINDEX_ARG_DROPPEDCOUNT_ARG_61F126B8
= "订阅队列已满，丢弃 entry，searchIndex={}，droppedCount={}";
+ public static final String
LOG_SUBSCRIPTION_REALTIME_ADMISSION_REJECTED_ARG_ENTRY_S_IN_THE_LAST_ARG_MS_WAL_REPLAY_REQUIRED_GROUP_ARG_LATEST_SEARCHINDEX_ARG_REASONCODE_ARG_QUEUESIZE_ARG_QUEUEREMAINING_ARG_0FBF6226
=
+ "订阅实时准入拒绝了 {} 个 entry（最近 {} ms）；需通过 WAL 回放补齐，group={}，最新
searchIndex={}，reasonCode={}，queueSize={}，queueRemaining={}";
+ public static final String
LOG_SUBSCRIPTION_REALTIME_ADMISSION_REJECTED_ENTRY_WAL_REPLAY_REQUIRED_GROUP_ARG_SEARCHINDEX_ARG_REASONCODE_ARG_REJECTEDCOUNT_ARG_7F76D6A9
=
+ "订阅实时准入拒绝 entry；需通过 WAL
回放补齐，group={}，searchIndex={}，reasonCode={}，rejectedCount={}";
public static final String
LOG_RESERVED_ARG_BYTES_BATCH_ARG_ARG_CURRENT_TOTAL_USAGE_ARG_308AE9C2 = "预留 {}
字节给批次 {}-{}，当前总使用量 {}";
public static final String
LOG_ARG_FAILED_SEND_IDLE_WRITER_SAFE_TIME_BARRIER_ARG_STATUS_AE047EAD = "{}：无法向
{} 发送 idle writer safe-time barrier。状态={}";
public static final String
LOG_ARG_WRITE_OPERATION_FAILED_SEARCHINDEX_ARG_CODE_ARG_SUBSCRIPTIONQUEUES_ARG_THIS_ARG_F4B17576
= "{}：写入操作失败。searchIndex: {}。Code: {}，订阅队列：{}，当前对象：{}";
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueAdmission.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueAdmission.java
new file mode 100644
index 00000000000..18c43fd7e16
--- /dev/null
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueAdmission.java
@@ -0,0 +1,25 @@
+/*
+ * 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.consensus.iot.subscription;
+
+/** Exposes the reason for the calling thread's most recent subscription queue
offer rejection. */
+public interface SubscriptionQueueAdmission {
+
+ SubscriptionQueueRejectionReason getLastRejectionReason();
+}
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRegistry.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRegistry.java
index d63e42f12c9..0cc06eeff52 100644
---
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRegistry.java
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRegistry.java
@@ -32,20 +32,22 @@ import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicLong;
import java.util.function.LongSupplier;
public class SubscriptionQueueRegistry {
private static final Logger LOGGER =
LoggerFactory.getLogger(SubscriptionQueueRegistry.class);
- private static final long QUEUE_FULL_LOG_INTERVAL_MS =
TimeUnit.SECONDS.toMillis(10);
+ private static final long REJECTION_LOG_INTERVAL_MS =
TimeUnit.SECONDS.toMillis(10);
private final String consensusGroupId;
private final Map<BlockingQueue<IndexedConsensusRequest>,
SubscriptionQueueRegistration> queues =
new ConcurrentHashMap<>();
- private final AtomicLong droppedEntries = new AtomicLong();
- private final AtomicLong lastDropLogTimeMs = new AtomicLong();
+ // offer is synchronized, so these counts remain scoped to one reason
without extra atomics.
+ private final long[] rejectedEntriesByReason =
+ new long[SubscriptionQueueRejectionReason.values().length];
+ private final long[] lastRejectionLogTimeMsByReason =
+ new long[SubscriptionQueueRejectionReason.values().length];
public SubscriptionQueueRegistry(final String consensusGroupId) {
this.consensusGroupId = consensusGroupId;
@@ -141,27 +143,34 @@ public class SubscriptionQueueRegistry {
queue.remainingCapacity());
}
if (!offered) {
- final long droppedCount = droppedEntries.incrementAndGet();
+ final SubscriptionQueueRejectionReason rejectionReason =
+ queue instanceof SubscriptionQueueAdmission
+ ? ((SubscriptionQueueAdmission) queue).getLastRejectionReason()
+ : SubscriptionQueueRejectionReason.QUEUE_CAPACITY;
+ final int reasonIndex = rejectionReason.ordinal();
+ final long rejectedCount = ++rejectedEntriesByReason[reasonIndex];
final long now = System.currentTimeMillis();
- final long lastLogTime = lastDropLogTimeMs.get();
- if (now - lastLogTime >= QUEUE_FULL_LOG_INTERVAL_MS
- && lastDropLogTimeMs.compareAndSet(lastLogTime, now)) {
+ if (now - lastRejectionLogTimeMsByReason[reasonIndex] >=
REJECTION_LOG_INTERVAL_MS) {
+ lastRejectionLogTimeMsByReason[reasonIndex] = now;
+ rejectedEntriesByReason[reasonIndex] = 0L;
LOGGER.warn(
IoTConsensusMessages
-
.LOG_SUBSCRIPTION_QUEUE_FULL_DROPPED_ARG_ENTRY_S_LAST_ARG_MS_2AD8AB3D
- + IoTConsensusMessages
-
.LOG_SEARCHINDEX_ARG_QUEUESIZE_ARG_QUEUEREMAINING_ARG_2EA619ED,
- droppedEntries.getAndSet(0),
- QUEUE_FULL_LOG_INTERVAL_MS,
+
.LOG_SUBSCRIPTION_REALTIME_ADMISSION_REJECTED_ARG_ENTRY_S_IN_THE_LAST_ARG_MS_WAL_REPLAY_REQUIRED_GROUP_ARG_LATEST_SEARCHINDEX_ARG_REASONCODE_ARG_QUEUESIZE_ARG_QUEUEREMAINING_ARG_0FBF6226,
+ rejectedCount,
+ REJECTION_LOG_INTERVAL_MS,
+ consensusGroupId,
indexedConsensusRequest.getSearchIndex(),
+ rejectionReason.getCode(),
queue.size(),
queue.remainingCapacity());
} else if (LOGGER.isDebugEnabled()) {
LOGGER.debug(
IoTConsensusMessages
-
.LOG_SUBSCRIPTION_QUEUE_FULL_DROPPED_ENTRY_SEARCHINDEX_ARG_DROPPEDCOUNT_ARG_61F126B8,
+
.LOG_SUBSCRIPTION_REALTIME_ADMISSION_REJECTED_ENTRY_WAL_REPLAY_REQUIRED_GROUP_ARG_SEARCHINDEX_ARG_REASONCODE_ARG_REJECTEDCOUNT_ARG_7F76D6A9,
+ consensusGroupId,
indexedConsensusRequest.getSearchIndex(),
- droppedCount);
+ rejectionReason.getCode(),
+ rejectedCount);
}
}
}
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRejectionReason.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRejectionReason.java
new file mode 100644
index 00000000000..852bc55e2f6
--- /dev/null
+++
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/subscription/SubscriptionQueueRejectionReason.java
@@ -0,0 +1,42 @@
+/*
+ * 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.consensus.iot.subscription;
+
+/** Stable error codes used when a consensus subscription queue rejects
realtime input. */
+public enum SubscriptionQueueRejectionReason {
+ NONE("NONE"),
+ QUEUE_CAPACITY("SUBSCRIPTION_QUEUE_CAPACITY"),
+ SUBSCRIPTION_MEMORY_QUOTA("SUBSCRIPTION_MEMORY_QUOTA"),
+ SUBSCRIPTION_MEMORY_LIMIT("SUBSCRIPTION_MEMORY_LIMIT"),
+ SUBSCRIPTION_OVERSIZED_ENTRY("SUBSCRIPTION_OVERSIZED_ENTRY"),
+ WRITER_BACKLOG("SUBSCRIPTION_WRITER_BACKLOG"),
+ SEEK_IN_PROGRESS("SUBSCRIPTION_SEEK_IN_PROGRESS"),
+ CONSENSUS_REQUEST_MEMORY_LIMIT("CONSENSUS_REQUEST_MEMORY_LIMIT"),
+ INACTIVE_OR_CLOSED("SUBSCRIPTION_QUEUE_INACTIVE_OR_CLOSED");
+
+ private final String code;
+
+ SubscriptionQueueRejectionReason(final String code) {
+ this.code = code;
+ }
+
+ public String getCode() {
+ return code;
+ }
+}
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 5e92afb5f9c..3cd7ae14585 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
@@ -2644,4 +2644,6 @@ public final class DataNodePipeMessages {
"Interrupted while reading OPC UA server operation limits, use defaults:
maxNodesPerWrite={}, maxNodesPerNodeManagement={}";
public static final String
LOG_FAILED_TO_READ_OPC_UA_SERVER_OPERATION_LIMITS_USE_DEFAULTS_MAXNODESPERWRITE_ARG_MAXNODESPERNODEMANAGEMENT_ARG_65460871
=
"Failed to read OPC UA server operation limits, use defaults:
maxNodesPerWrite={}, maxNodesPerNodeManagement={}";
+ public static final String
MESSAGE_ARG_SUBSCRIPTION_ENTRY_REQUIRES_ARG_BYTES_ABOVE_THE_CURRENT_PER_QUEUE_MAXIMUM_ARG_BYTES_DATANODE_BUDGET_ARG_BYTES_REDUCE_THE_WRITE_BATCH_FIELD_SIZE_OR_INCREASE_SUBSCRIPTION_MATERIALIZATION_MEMORY_WAL_PROGRESS_HAS_NOT_ADVANCED_AFCBC7FC
=
+ "[%s] Subscription entry requires %d bytes, above the current per-queue
maximum %d bytes (DataNode budget %d bytes). Reduce the write batch/field size
or increase subscription materialization memory. WAL progress has not
advanced.";
}
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 14bc9cd4b26..23f58a00c51 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
@@ -2469,4 +2469,6 @@ public final class DataNodePipeMessages {
"读取 OPC UA
服务器操作限制时被中断，使用默认值：maxNodesPerWrite={}，maxNodesPerNodeManagement={}";
public static final String
LOG_FAILED_TO_READ_OPC_UA_SERVER_OPERATION_LIMITS_USE_DEFAULTS_MAXNODESPERWRITE_ARG_MAXNODESPERNODEMANAGEMENT_ARG_65460871
=
"读取 OPC UA
服务器操作限制失败，使用默认值：maxNodesPerWrite={}，maxNodesPerNodeManagement={}";
+ public static final String
MESSAGE_ARG_SUBSCRIPTION_ENTRY_REQUIRES_ARG_BYTES_ABOVE_THE_CURRENT_PER_QUEUE_MAXIMUM_ARG_BYTES_DATANODE_BUDGET_ARG_BYTES_REDUCE_THE_WRITE_BATCH_FIELD_SIZE_OR_INCREASE_SUBSCRIPTION_MATERIALIZATION_MEMORY_WAL_PROGRESS_HAS_NOT_ADVANCED_AFCBC7FC
=
+ "[%s] 订阅 entry 需要 %d 字节，超过当前队列可用上限 %d 字节（DataNode 预算 %d 字节）。请减小写入
batch/字段大小或增加订阅物化内存。WAL 进度未推进。";
}
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 22fd01f6c47..b8cd6b258d3 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
@@ -30,6 +30,8 @@ import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl;
import org.apache.iotdb.consensus.iot.SubscriptionWalRetentionPolicy;
import org.apache.iotdb.consensus.iot.log.ConsensusReqReader;
import org.apache.iotdb.consensus.iot.logdispatcher.IoTConsensusMemoryManager;
+import org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueAdmission;
+import
org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueRejectionReason;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
@@ -92,10 +94,11 @@ import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicLongArray;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.BiConsumer;
-import java.util.function.BooleanSupplier;
+import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.function.Supplier;
@@ -151,6 +154,27 @@ public class ConsensusPrefetchingQueue {
private SubscriptionMemoryManager subscriptionMemoryManager;
+ /** Queue-local view of the DataNode subscription memory budget. */
+ private SubscriptionMemoryManager.MemoryHandle subscriptionMemoryHandle;
+
+ private final AtomicLong realtimeAdmissionRejectionCount = new
AtomicLong(0L);
+
+ private final AtomicLongArray realtimeAdmissionRejectionsByReason =
+ new AtomicLongArray(SubscriptionQueueRejectionReason.values().length);
+
+ private final AtomicLong subscriptionMemoryRejectionCount = new
AtomicLong(0L);
+
+ private final AtomicLong oversizedEntryRejectionCount = new AtomicLong(0L);
+
+ private final AtomicReference<SubscriptionQueueRejectionReason>
lastRejectionReason =
+ new AtomicReference<>(SubscriptionQueueRejectionReason.NONE);
+
+ private volatile SubscriptionQueueRejectionReason
realtimeAdmissionBlockReason =
+ SubscriptionQueueRejectionReason.NONE;
+
+ private volatile SubscriptionQueueRejectionReason memoryBlockReason =
+ SubscriptionQueueRejectionReason.NONE;
+
/**
* Incremented on each seek to distinguish batches/events created before and
after the seek. The
* value is copied into commit contexts so stale events cannot be committed
after a later seek.
@@ -415,30 +439,48 @@ public class ConsensusPrefetchingQueue {
}
private static final class WakeableIndexedConsensusQueue
- extends LinkedBlockingDeque<IndexedConsensusRequest> {
+ extends LinkedBlockingDeque<IndexedConsensusRequest> implements
SubscriptionQueueAdmission {
private final Runnable wakeupHook;
- private final BooleanSupplier admissionSupplier;
+ private final Supplier<SubscriptionQueueRejectionReason> admissionSupplier;
+ private final Consumer<SubscriptionQueueRejectionReason> rejectionListener;
private final IoTConsensusMemoryManager requestMemoryManager =
IoTConsensusMemoryManager.getInstance();
private final AtomicLong retainedRequestBytes = new AtomicLong(0L);
+ // The registry reads this immediately after offer on the same writer
thread. A queue-wide
+ // value could be overwritten by another writer before the registry logs
the rejection.
+ private final ThreadLocal<SubscriptionQueueRejectionReason>
lastOfferRejectionReason =
+ ThreadLocal.withInitial(() -> SubscriptionQueueRejectionReason.NONE);
private WakeableIndexedConsensusQueue(
- final int capacity, final Runnable wakeupHook, final BooleanSupplier
admissionSupplier) {
+ final int capacity,
+ final Runnable wakeupHook,
+ final Supplier<SubscriptionQueueRejectionReason> admissionSupplier,
+ final Consumer<SubscriptionQueueRejectionReason> rejectionListener) {
super(capacity);
this.wakeupHook = wakeupHook;
this.admissionSupplier = admissionSupplier;
+ this.rejectionListener = rejectionListener;
+ }
+
+ private boolean reject(final SubscriptionQueueRejectionReason reason) {
+ final SubscriptionQueueRejectionReason effectiveReason =
+ Objects.isNull(reason) ?
SubscriptionQueueRejectionReason.QUEUE_CAPACITY : reason;
+ lastOfferRejectionReason.set(effectiveReason);
+ rejectionListener.accept(effectiveReason);
+ return false;
}
@Override
public boolean offer(final IndexedConsensusRequest request) {
final boolean offered;
synchronized (this) {
- if (!admissionSupplier.getAsBoolean()) {
- return false;
+ final SubscriptionQueueRejectionReason admissionReason =
admissionSupplier.get();
+ if (admissionReason != SubscriptionQueueRejectionReason.NONE) {
+ return reject(admissionReason);
}
if (!requestMemoryManager.reserve(request)) {
- return false;
+ return
reject(SubscriptionQueueRejectionReason.CONSENSUS_REQUEST_MEMORY_LIMIT);
}
try {
offered = super.offer(request);
@@ -450,6 +492,7 @@ public class ConsensusPrefetchingQueue {
retainedRequestBytes.addAndGet(request.getRetainedMemorySize());
} else {
requestMemoryManager.free(request);
+ reject(SubscriptionQueueRejectionReason.QUEUE_CAPACITY);
}
}
if (offered) {
@@ -478,6 +521,11 @@ public class ConsensusPrefetchingQueue {
private long getRetainedRequestBytes() {
return retainedRequestBytes.get();
}
+
+ @Override
+ public SubscriptionQueueRejectionReason getLastRejectionReason() {
+ return lastOfferRejectionReason.get();
+ }
}
private static final class PendingSeekRequest {
@@ -559,6 +607,8 @@ public class ConsensusPrefetchingQueue {
this.tableModel = converter.isTableModel();
this.commitManager = commitManager;
this.subscriptionMemoryManager =
SubscriptionDataNodeResourceManager.memory();
+ this.subscriptionMemoryHandle = subscriptionMemoryManager.registerQueue();
+ this.subscriptionMemoryHandle.setActive(initialActive);
this.fallbackCommittedRegionProgress = fallbackCommittedRegionProgress;
this.fallbackTailSearchIndex = tailStartSearchIndex;
this.runtimeVersion = initialRuntimeVersion;
@@ -577,7 +627,8 @@ public class ConsensusPrefetchingQueue {
new WakeableIndexedConsensusQueue(
PENDING_QUEUE_CAPACITY,
this::requestPrefetchForRealtimeEntry,
- this::canAcceptRealtimeEntry);
+ this::getRealtimeAdmissionRejectionReason,
+ this::recordRealtimeAdmissionRejection);
serverImpl.registerSubscriptionQueue(
pendingEntries, retentionPolicy,
this::getRequiredRetainedMinVersionId);
@@ -602,7 +653,12 @@ public class ConsensusPrefetchingQueue {
if (retainedTabletBytes.get() != 0L) {
throw new IllegalStateException();
}
+ if (Objects.nonNull(this.subscriptionMemoryHandle)) {
+ this.subscriptionMemoryHandle.close();
+ }
this.subscriptionMemoryManager =
Objects.requireNonNull(subscriptionMemoryManager);
+ this.subscriptionMemoryHandle =
this.subscriptionMemoryManager.registerQueue();
+ this.subscriptionMemoryHandle.setActive(isActive);
memoryBlockedEntryBytes = -1L;
realtimeAdmissionBlocked.set(false);
}
@@ -641,23 +697,53 @@ public class ConsensusPrefetchingQueue {
}
}
- private boolean canAcceptRealtimeEntry() {
- return isActive
- && !closeRequested
- && !isClosed
- && !realtimeAdmissionBlocked.get()
- && subscriptionMemoryManager.getFreeMemorySizeInBytes() > 0L;
+ private SubscriptionQueueRejectionReason
getRealtimeAdmissionRejectionReason() {
+ if (!isActive || closeRequested || isClosed) {
+ return SubscriptionQueueRejectionReason.INACTIVE_OR_CLOSED;
+ }
+ if (realtimeAdmissionBlocked.get()) {
+ return realtimeAdmissionBlockReason;
+ }
+ // Allow an empty queue to probe an entry even with a zero quota so
materialization can report
+ // an explicit oversized-entry error. Pending requests have separate
consensus memory
+ // accounting.
+ if (subscriptionMemoryHandle.getUsedMemorySizeInBytes() > 0L
+ && subscriptionMemoryHandle.getFreeMemorySizeInBytes() <= 0L) {
+ return
toQueueRejectionReason(subscriptionMemoryHandle.inspectRejection(1L));
+ }
+ return SubscriptionQueueRejectionReason.NONE;
}
- private void blockRealtimeAdmission() {
+ private void blockRealtimeAdmission(final SubscriptionQueueRejectionReason
reason) {
+ realtimeAdmissionBlockReason = reason;
realtimeAdmissionBlocked.set(true);
- pendingEntries.clear();
}
private void unblockRealtimeAdmission() {
realtimeAdmissionBlocked.set(false);
}
+ private void recordRealtimeAdmissionRejection(
+ final SubscriptionQueueRejectionReason rejectionReason) {
+ if (Objects.isNull(rejectionReason)
+ || rejectionReason == SubscriptionQueueRejectionReason.NONE) {
+ return;
+ }
+ lastRejectionReason.set(rejectionReason);
+ realtimeAdmissionRejectionCount.incrementAndGet();
+
realtimeAdmissionRejectionsByReason.incrementAndGet(rejectionReason.ordinal());
+ recordMemoryRejection(rejectionReason);
+ }
+
+ private void recordMemoryRejection(final SubscriptionQueueRejectionReason
rejectionReason) {
+ if (rejectionReason ==
SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_QUOTA
+ || rejectionReason ==
SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_LIMIT) {
+ subscriptionMemoryRejectionCount.incrementAndGet();
+ } else if (rejectionReason ==
SubscriptionQueueRejectionReason.SUBSCRIPTION_OVERSIZED_ENTRY) {
+ oversizedEntryRejectionCount.incrementAndGet();
+ }
+ }
+
private ConsensusPrefetchSubtask ensurePrefetchSubtaskBound() {
if (closeRequested || isClosed) {
return null;
@@ -777,6 +863,26 @@ public class ConsensusPrefetchingQueue {
} else if (Objects.isNull(event) &&
shouldRecoverPrefetchBindingAfterEmptyPoll()) {
requestPrefetch();
}
+ // Deliver reserved events first so ACK can drain memory. An entry that
cannot fit even in an
+ // empty queue is an explicit configuration error, with no replay/commit
progress advancement.
+ final long blockedEntryBytes = memoryBlockedEntryBytes;
+ if (Objects.isNull(event)
+ && blockedEntryBytes > 0L
+ && subscriptionMemoryHandle.inspectRejection(blockedEntryBytes)
+ ==
SubscriptionMemoryManager.AllocationRejectionReason.OVERSIZED_ENTRY) {
+ return new SubscriptionEvent(
+ SubscriptionPollResponseType.ERROR.getType(),
+ new ErrorPayload(
+ String.format(
+ DataNodePipeMessages
+
.MESSAGE_ARG_SUBSCRIPTION_ENTRY_REQUIRES_ARG_BYTES_ABOVE_THE_CURRENT_PER_QUEUE_MAXIMUM_ARG_BYTES_DATANODE_BUDGET_ARG_BYTES_REDUCE_THE_WRITE_BATCH_FIELD_SIZE_OR_INCREASE_SUBSCRIPTION_MATERIALIZATION_MEMORY_WAL_PROGRESS_HAS_NOT_ADVANCED_AFCBC7FC,
+
SubscriptionQueueRejectionReason.SUBSCRIPTION_OVERSIZED_ENTRY.getCode(),
+ blockedEntryBytes,
+ subscriptionMemoryHandle.getMaximumMemorySizeInBytes(),
+ subscriptionMemoryManager.getTotalMemorySizeInBytes()),
+ true),
+
createNonCommittableContext(IoTDBDescriptor.getInstance().getConfig().getDataNodeId()));
+ }
return event;
} finally {
releaseReadLock();
@@ -1434,7 +1540,7 @@ public class ConsensusPrefetchingQueue {
recycleInFlightEvents();
if (!isActive) {
- blockRealtimeAdmission();
+
blockRealtimeAdmission(SubscriptionQueueRejectionReason.INACTIVE_OR_CLOSED);
return computeIdleRoundResult();
}
if (prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE) {
@@ -1457,12 +1563,12 @@ public class ConsensusPrefetchingQueue {
if (prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE
|| !realtimeEntriesByWriter.isEmpty()) {
if (!realtimeEntriesByWriter.isEmpty()) {
- blockRealtimeAdmission();
+
blockRealtimeAdmission(SubscriptionQueueRejectionReason.WRITER_BACKLOG);
}
return computeIdleRoundResult();
}
if (shouldWaitForSubscriptionMemory()) {
- blockRealtimeAdmission();
+ blockRealtimeAdmission(memoryBlockReason);
if (!lingerBatch.isEmpty() && !flushBatch(lingerBatch,
observedSeekGeneration)) {
resetRoundStateForSeek(seekGeneration.get());
return PrefetchRoundResult.rescheduleNow();
@@ -1507,7 +1613,7 @@ public class ConsensusPrefetchingQueue {
if (batchResult == MaterializationResult.MEMORY_BLOCKED) {
// Publish already reserved Tablets immediately so ACK can release
their memory. The
// blocked request and the drained suffix are recovered from WAL
on a later round.
- blockRealtimeAdmission();
+ blockRealtimeAdmission(memoryBlockReason);
if (!lingerBatch.isEmpty() && !flushBatch(lingerBatch,
observedSeekGeneration)) {
resetRoundStateForSeek(seekGeneration.get());
return PrefetchRoundResult.rescheduleNow();
@@ -1522,7 +1628,7 @@ public class ConsensusPrefetchingQueue {
if (batch.isEmpty() && lingerBatch.isEmpty()) {
final MaterializationResult walResult =
tryCatchUpFromWAL(observedSeekGeneration);
if (walResult == MaterializationResult.MEMORY_BLOCKED) {
- blockRealtimeAdmission();
+ blockRealtimeAdmission(memoryBlockReason);
return PrefetchRoundResult.rescheduleAfter(MEMORY_RETRY_SLEEP_MS);
}
if (walResult == MaterializationResult.STALE) {
@@ -2327,11 +2433,19 @@ public class ConsensusPrefetchingQueue {
memoryBlockedEntryBytes = -1L;
return true;
}
- if (!subscriptionMemoryManager.tryAllocate(bytes)) {
+ final SubscriptionMemoryManager.AllocationResult allocationResult =
+ subscriptionMemoryHandle.tryAllocate(bytes);
+ if (!allocationResult.isAccepted()) {
memoryBlockedEntryBytes = bytes;
+ final SubscriptionQueueRejectionReason rejectionReason =
+ toQueueRejectionReason(allocationResult.getRejectionReason());
+ memoryBlockReason = rejectionReason;
+ lastRejectionReason.set(rejectionReason);
+ recordMemoryRejection(rejectionReason);
return false;
}
memoryBlockedEntryBytes = -1L;
+ memoryBlockReason = SubscriptionQueueRejectionReason.NONE;
retainedTabletBytes.addAndGet(bytes);
return true;
}
@@ -2339,22 +2453,21 @@ public class ConsensusPrefetchingQueue {
private boolean shouldWaitForSubscriptionMemory() {
final long blockedEntryBytes = memoryBlockedEntryBytes;
if (blockedEntryBytes <= 0L) {
- return subscriptionMemoryManager.getFreeMemorySizeInBytes() <= 0L;
- }
-
- final long totalMemorySizeInBytes =
subscriptionMemoryManager.getTotalMemorySizeInBytes();
- final long usedMemorySizeInBytes =
subscriptionMemoryManager.getUsedMemorySizeInBytes();
- final boolean canAllocateWithinLimit =
- blockedEntryBytes <= totalMemorySizeInBytes
- && totalMemorySizeInBytes - usedMemorySizeInBytes >=
blockedEntryBytes;
- final boolean canAllocateSingleOversizedEntry =
- totalMemorySizeInBytes > 0L
- && usedMemorySizeInBytes == 0L
- && blockedEntryBytes > totalMemorySizeInBytes;
- if (canAllocateWithinLimit || canAllocateSingleOversizedEntry) {
+ if (subscriptionMemoryHandle.getUsedMemorySizeInBytes() > 0L
+ && subscriptionMemoryHandle.getFreeMemorySizeInBytes() <= 0L) {
+ memoryBlockReason =
toQueueRejectionReason(subscriptionMemoryHandle.inspectRejection(1L));
+ return memoryBlockReason != SubscriptionQueueRejectionReason.NONE;
+ }
+ return false;
+ }
+ final SubscriptionMemoryManager.AllocationRejectionReason rejectionReason =
+ subscriptionMemoryHandle.inspectRejection(blockedEntryBytes);
+ if (rejectionReason ==
SubscriptionMemoryManager.AllocationRejectionReason.NONE) {
memoryBlockedEntryBytes = -1L;
+ memoryBlockReason = SubscriptionQueueRejectionReason.NONE;
return false;
}
+ memoryBlockReason = toQueueRejectionReason(rejectionReason);
return true;
}
@@ -2372,7 +2485,22 @@ public class ConsensusPrefetchingQueue {
releasedBytes = Math.min(bytes, currentRetainedBytes);
} while (!retainedTabletBytes.compareAndSet(
currentRetainedBytes, currentRetainedBytes - releasedBytes));
- subscriptionMemoryManager.release(releasedBytes);
+ subscriptionMemoryHandle.release(releasedBytes);
+ }
+
+ private SubscriptionQueueRejectionReason toQueueRejectionReason(
+ final SubscriptionMemoryManager.AllocationRejectionReason
rejectionReason) {
+ switch (rejectionReason) {
+ case NONE:
+ return SubscriptionQueueRejectionReason.NONE;
+ case MEMORY_QUOTA:
+ return SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_QUOTA;
+ case OVERSIZED_ENTRY:
+ return SubscriptionQueueRejectionReason.SUBSCRIPTION_OVERSIZED_ENTRY;
+ case MEMORY_LIMIT:
+ default:
+ return SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_LIMIT;
+ }
}
private void reconcileRetainedTabletMemoryAfterCleanup() {
@@ -3038,7 +3166,7 @@ public class ConsensusPrefetchingQueue {
// ======================== Cleanup ========================
public void cleanUp() {
- blockRealtimeAdmission();
+
blockRealtimeAdmission(SubscriptionQueueRejectionReason.INACTIVE_OR_CLOSED);
acquireWriteLock();
try {
prefetchingQueue.forEach(event -> cleanUpEvent(event, true));
@@ -3054,7 +3182,8 @@ public class ConsensusPrefetchingQueue {
materializedProgressByWriter.clear();
// A prefetch round that already held the read lock may have reopened
admission while this
// cleanup was waiting for the write lock. Fence and clear once more
under the write lock.
- blockRealtimeAdmission();
+
blockRealtimeAdmission(SubscriptionQueueRejectionReason.INACTIVE_OR_CLOSED);
+ pendingEntries.clear();
discardBatch(lingerBatch);
reconcileRetainedTabletMemoryAfterCleanup();
memoryBlockedEntryBytes = -1L;
@@ -3065,6 +3194,8 @@ public class ConsensusPrefetchingQueue {
pendingSubscriptionWalResetSearchIndex = Long.MIN_VALUE;
pendingSubscriptionWalResetGeneration = Long.MIN_VALUE;
closeSubscriptionWALIterator();
+ subscriptionMemoryHandle.setActive(false);
+ subscriptionMemoryHandle.close();
} finally {
releaseWriteLock();
@@ -3323,7 +3454,7 @@ public class ConsensusPrefetchingQueue {
}
private void applySeekResetUnderWriteLock(final PendingSeekRequest request) {
- blockRealtimeAdmission();
+ blockRealtimeAdmission(SubscriptionQueueRejectionReason.SEEK_IN_PROGRESS);
// 1. Clean up all queued and in-flight events
prefetchingQueue.forEach(event -> cleanUpEvent(event, true));
@@ -3333,6 +3464,7 @@ public class ConsensusPrefetchingQueue {
discardRetainedEventMemory();
// 2. Discard stale pending entries from in-memory queue
+ pendingEntries.clear();
memoryBlockedEntryBytes = -1L;
unblockRealtimeAdmission();
@@ -4044,12 +4176,14 @@ public class ConsensusPrefetchingQueue {
if (this.isActive != active) {
if (active) {
this.isActive = true;
+ subscriptionMemoryHandle.setActive(true);
unblockRealtimeAdmission();
} else {
// Fence admission before waiting for the write lock so no new
realtime reference can race
// with the lifecycle cleanup below.
this.isActive = false;
- realtimeAdmissionBlocked.set(true);
+ subscriptionMemoryHandle.setActive(false);
+
blockRealtimeAdmission(SubscriptionQueueRejectionReason.INACTIVE_OR_CLOSED);
acquireWriteLock();
try {
seekRequestToFail = pendingSeekRequest;
@@ -4400,6 +4534,39 @@ public class ConsensusPrefetchingQueue {
return subscriptionMemoryManager.getTotalMemorySizeInBytes();
}
+ public long getSubscriptionMemoryUsedInBytes() {
+ return subscriptionMemoryHandle.getUsedMemorySizeInBytes();
+ }
+
+ public long getSubscriptionMemoryQuotaInBytes() {
+ return subscriptionMemoryHandle.getMemoryQuotaInBytes();
+ }
+
+ public long getSubscriptionMemoryMaximumInBytes() {
+ return subscriptionMemoryHandle.getMaximumMemorySizeInBytes();
+ }
+
+ public long getRealtimeAdmissionRejectionCount() {
+ return realtimeAdmissionRejectionCount.get();
+ }
+
+ public long getRealtimeAdmissionRejectionCount(
+ final SubscriptionQueueRejectionReason rejectionReason) {
+ return realtimeAdmissionRejectionsByReason.get(rejectionReason.ordinal());
+ }
+
+ public long getSubscriptionMemoryRejectionCount() {
+ return subscriptionMemoryRejectionCount.get();
+ }
+
+ public long getOversizedEntryRejectionCount() {
+ return oversizedEntryRejectionCount.get();
+ }
+
+ public String getLastAdmissionRejectionCode() {
+ return lastRejectionReason.get().getCode();
+ }
+
/** Exposes the current seek generation for runtime tests and metrics. */
public long getCurrentSeekGeneration() {
return seekGeneration.get();
@@ -4477,6 +4644,17 @@ public class ConsensusPrefetchingQueue {
result.put(
"subscriptionMemoryLimitInBytes",
String.valueOf(subscriptionMemoryManager.getTotalMemorySizeInBytes()));
+ result.put("subscriptionMemoryUsedInBytes",
String.valueOf(getSubscriptionMemoryUsedInBytes()));
+ result.put(
+ "subscriptionMemoryQuotaInBytes",
String.valueOf(getSubscriptionMemoryQuotaInBytes()));
+ result.put(
+ "subscriptionMemoryMaximumInBytes",
String.valueOf(getSubscriptionMemoryMaximumInBytes()));
+ result.put(
+ "realtimeAdmissionRejectionCount",
String.valueOf(getRealtimeAdmissionRejectionCount()));
+ result.put(
+ "subscriptionMemoryRejectionCount",
String.valueOf(getSubscriptionMemoryRejectionCount()));
+ result.put("oversizedEntryRejectionCount",
String.valueOf(getOversizedEntryRejectionCount()));
+ result.put("lastAdmissionRejectionCode", getLastAdmissionRejectionCode());
result.put("pendingPathAcceptedEntries",
String.valueOf(getPendingPathAcceptedEntries()));
result.put("walPathAcceptedEntries",
String.valueOf(getWalPathAcceptedEntries()));
result.put("seekGeneration", String.valueOf(seekGeneration.get()));
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..0a549565b06 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
@@ -21,6 +21,7 @@ package org.apache.iotdb.db.subscription.metric;
import org.apache.iotdb.commons.service.metric.enums.Metric;
import org.apache.iotdb.commons.service.metric.enums.Tag;
+import
org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueRejectionReason;
import org.apache.iotdb.db.i18n.DataNodePipeMessages;
import
org.apache.iotdb.db.subscription.broker.consensus.ConsensusPrefetchingQueue;
import org.apache.iotdb.metrics.AbstractMetricService;
@@ -199,6 +200,53 @@ public class ConsensusSubscriptionPrefetchingQueueMetrics
implements IMetricSet
queue,
ConsensusPrefetchingQueue::getProgressStatus,
key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_MEMORY_USED.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getSubscriptionMemoryUsedInBytes,
+ key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_MEMORY_QUOTA.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getSubscriptionMemoryQuotaInBytes,
+ key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_MEMORY_MAXIMUM.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getSubscriptionMemoryMaximumInBytes,
+ key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_ADMISSION_REJECTIONS.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getRealtimeAdmissionRejectionCount,
+ key.getTags());
+ for (final SubscriptionQueueRejectionReason reason :
+ SubscriptionQueueRejectionReason.values()) {
+ if (reason != SubscriptionQueueRejectionReason.NONE) {
+ metricService.createAutoGauge(
+
Metric.SUBSCRIPTION_CONSENSUS_ADMISSION_REJECTIONS_BY_REASON.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ q -> q.getRealtimeAdmissionRejectionCount(reason),
+ key.getTagsWithReason(reason));
+ }
+ }
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_MEMORY_REJECTIONS.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getSubscriptionMemoryRejectionCount,
+ key.getTags());
+ metricService.createAutoGauge(
+ Metric.SUBSCRIPTION_CONSENSUS_OVERSIZED_REJECTIONS.toString(),
+ MetricLevel.IMPORTANT,
+ queue,
+ ConsensusPrefetchingQueue::getOversizedEntryRejectionCount,
+ key.getTags());
}
private void createRate(final QueueMetricsKey key) {
@@ -309,6 +357,37 @@ public class ConsensusSubscriptionPrefetchingQueueMetrics
implements IMetricSet
MetricType.AUTO_GAUGE,
Metric.SUBSCRIPTION_CONSENSUS_PROGRESS_STATUS.toString(),
key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
Metric.SUBSCRIPTION_CONSENSUS_MEMORY_USED.toString(), key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.SUBSCRIPTION_CONSENSUS_MEMORY_QUOTA.toString(),
+ key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.SUBSCRIPTION_CONSENSUS_MEMORY_MAXIMUM.toString(),
+ key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.SUBSCRIPTION_CONSENSUS_ADMISSION_REJECTIONS.toString(),
+ key.getTags());
+ for (final SubscriptionQueueRejectionReason reason :
+ SubscriptionQueueRejectionReason.values()) {
+ if (reason != SubscriptionQueueRejectionReason.NONE) {
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+
Metric.SUBSCRIPTION_CONSENSUS_ADMISSION_REJECTIONS_BY_REASON.toString(),
+ key.getTagsWithReason(reason));
+ }
+ }
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.SUBSCRIPTION_CONSENSUS_MEMORY_REJECTIONS.toString(),
+ key.getTags());
+ metricService.remove(
+ MetricType.AUTO_GAUGE,
+ Metric.SUBSCRIPTION_CONSENSUS_OVERSIZED_REJECTIONS.toString(),
+ key.getTags());
}
private void removeRate(final QueueMetricsKey key) {
@@ -357,6 +436,17 @@ public class ConsensusSubscriptionPrefetchingQueueMetrics
implements IMetricSet
};
}
+ private String[] getTagsWithReason(final SubscriptionQueueRejectionReason
reason) {
+ return new String[] {
+ Tag.NAME.toString(),
+ queueId,
+ Tag.REGION.toString(),
+ regionId,
+ Tag.REASON.toString(),
+ reason.getCode(),
+ };
+ }
+
@Override
public boolean equals(final Object obj) {
if (this == obj) {
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..d250cd0ba16 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
@@ -25,11 +25,29 @@ import org.apache.iotdb.commons.memory.MemoryBlockType;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * DataNode-wide memory manager for materialized subscription data.
+ *
+ * <p>Consensus subscription queues use a handle rather than the legacy
manager-level allocation
+ * methods. Handles are accounted independently. Each active queue keeps half
of its equal share
+ * protected; another queue may borrow only the remaining idle space. Borrowed
allocations are
+ * drained rather than revoked when the active queue set changes. An entry
larger than the maximum
+ * amount one queue can hold under this policy is rejected explicitly.
+ */
public class SubscriptionMemoryManager {
private static final String MEMORY_BLOCK_NAME = "Subscription";
private final IMemoryBlock memoryBlock;
+ private static final int PROTECTED_SHARE_DIVISOR = 2;
+ private final Map<Long, MemoryHandle> handles = new HashMap<>();
+ private long nextHandleId;
+
+ // Kept for callers of the original manager-level API. It is deliberately
not a fair-share owner.
+ private long legacyUsedMemoryInBytes;
SubscriptionMemoryManager() {
memoryBlock =
@@ -46,44 +64,321 @@ public class SubscriptionMemoryManager {
MEMORY_BLOCK_NAME, null, totalMemorySizeInBytes,
MemoryBlockType.DYNAMIC);
}
- /**
- * Reserves memory for materialized subscription data.
- *
- * <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.
- */
+ /** Registers a queue for fair-share accounting. */
+ public synchronized MemoryHandle registerQueue() {
+ final MemoryHandle handle = new MemoryHandle(++nextHandleId);
+ handles.put(handle.id, handle);
+ return handle;
+ }
+
+ /** Reserves memory through the compatibility owner. New queues should use a
handle. */
public synchronized boolean tryAllocate(final long sizeInBytes) {
+ return tryAllocateLegacy(sizeInBytes).isAccepted();
+ }
+
+ private AllocationResult tryAllocateLegacy(final long sizeInBytes) {
if (sizeInBytes <= 0L) {
- return true;
+ return AllocationResult.accepted(getTotalMemorySizeInBytes(),
getFreeMemorySizeInBytes());
}
- if (memoryBlock.allocate(sizeInBytes)) {
- return true;
+ final long total = getTotalMemorySizeInBytes();
+ final long used = getUsedMemorySizeInBytes();
+ if (total > 0L && used <= total && sizeInBytes <= total - used) {
+ if (!memoryBlock.allocate(sizeInBytes)) {
+ return AllocationResult.rejected(
+ AllocationRejectionReason.MEMORY_LIMIT, total,
getFreeMemorySizeInBytes());
+ }
+ legacyUsedMemoryInBytes += sizeInBytes;
+ return AllocationResult.accepted(total, total - used - sizeInBytes);
}
- if (memoryBlock.getTotalMemorySizeInBytes() > 0L
- && memoryBlock.getUsedMemoryInBytes() == 0L
- && sizeInBytes > memoryBlock.getTotalMemorySizeInBytes()) {
+ if (total > 0L && used == 0L && sizeInBytes > total) {
memoryBlock.forceAllocateWithoutLimitation(sizeInBytes);
- return true;
+ legacyUsedMemoryInBytes += sizeInBytes;
+ return AllocationResult.accepted(total, total - sizeInBytes);
}
- return false;
+ return AllocationResult.rejected(
+ sizeInBytes > total
+ ? AllocationRejectionReason.OVERSIZED_ENTRY
+ : AllocationRejectionReason.MEMORY_LIMIT,
+ total,
+ total - used);
}
public synchronized void release(final long sizeInBytes) {
- if (sizeInBytes > 0L) {
- memoryBlock.release(sizeInBytes);
+ if (sizeInBytes <= 0L) {
+ return;
+ }
+ final long released = Math.min(sizeInBytes, legacyUsedMemoryInBytes);
+ if (released > 0L) {
+ legacyUsedMemoryInBytes -= released;
+ memoryBlock.release(released);
}
}
- public long getTotalMemorySizeInBytes() {
+ public synchronized long getTotalMemorySizeInBytes() {
return memoryBlock.getTotalMemorySizeInBytes();
}
- public long getUsedMemorySizeInBytes() {
+ public synchronized long getUsedMemorySizeInBytes() {
return memoryBlock.getUsedMemoryInBytes();
}
- public long getFreeMemorySizeInBytes() {
+ public synchronized long getFreeMemorySizeInBytes() {
return memoryBlock.getFreeMemoryInBytes();
}
+
+ private synchronized AllocationResult tryAllocate(
+ final MemoryHandle handle, final long sizeInBytes) {
+ final AllocationResult decision = inspectAllocationUnderLock(handle,
sizeInBytes);
+ if (!decision.isAccepted() || sizeInBytes <= 0L) {
+ return decision;
+ }
+
+ if (!memoryBlock.allocate(sizeInBytes)) {
+ return AllocationResult.rejected(
+ AllocationRejectionReason.MEMORY_LIMIT,
+ handle.getMemoryQuotaInBytesUnderLock(),
+ getFreeMemorySizeInBytes());
+ }
+ handle.usedMemoryInBytes += sizeInBytes;
+ return AllocationResult.accepted(
+ handle.getMemoryQuotaInBytesUnderLock(),
handle.getFreeMemorySizeInBytes());
+ }
+
+ private AllocationResult inspectAllocationUnderLock(
+ final MemoryHandle handle, final long sizeInBytes) {
+ if (sizeInBytes <= 0L) {
+ return AllocationResult.accepted(
+ handle.getMemoryQuotaInBytes(), handle.getFreeMemorySizeInBytes());
+ }
+ if (!handle.active || handle.closed || !handles.containsKey(handle.id)) {
+ return AllocationResult.rejected(
+ AllocationRejectionReason.MEMORY_LIMIT, getTotalMemorySizeInBytes(),
0L);
+ }
+
+ final long total = getTotalMemorySizeInBytes();
+ final long used = getUsedMemorySizeInBytes();
+ final long quota = handle.getMemoryQuotaInBytesUnderLock();
+ final long protectedShare = quota / PROTECTED_SHARE_DIVISOR;
+ final long maximum = handle.getMaximumMemorySizeInBytesUnderLock();
+ long reservedByOthers = 0L;
+ for (final MemoryHandle other : handles.values()) {
+ if (other != handle && !other.closed && (other.active ||
other.usedMemoryInBytes > 0L)) {
+ reservedByOthers += Math.max(protectedShare, other.usedMemoryInBytes);
+ }
+ }
+ final long availableToOwner = Math.max(0L, total - legacyUsedMemoryInBytes
- reservedByOthers);
+ final long ownerUsed = handle.usedMemoryInBytes;
+ final boolean fitsInNode = total > 0L && used <= total && sizeInBytes <=
total - used;
+ final boolean fitsInShare =
+ ownerUsed <= availableToOwner && sizeInBytes <= availableToOwner -
ownerUsed;
+
+ if (fitsInNode && fitsInShare) {
+ return AllocationResult.accepted(quota, availableToOwner - ownerUsed);
+ }
+
+ final AllocationRejectionReason reason;
+ if (sizeInBytes > maximum) {
+ reason = AllocationRejectionReason.OVERSIZED_ENTRY;
+ } else if (ownerUsed > maximum || sizeInBytes > maximum - ownerUsed) {
+ reason = AllocationRejectionReason.MEMORY_QUOTA;
+ } else {
+ reason = AllocationRejectionReason.MEMORY_LIMIT;
+ }
+ return AllocationResult.rejected(
+ reason, quota, Math.min(availableToOwner - ownerUsed, total - used));
+ }
+
+ private synchronized void release(final MemoryHandle handle, final long
sizeInBytes) {
+ if (sizeInBytes <= 0L || handle.closed) {
+ return;
+ }
+ final long released = Math.min(sizeInBytes, handle.usedMemoryInBytes);
+ if (released > 0L) {
+ handle.usedMemoryInBytes -= released;
+ memoryBlock.release(released);
+ }
+ }
+
+ private synchronized void unregister(final MemoryHandle handle) {
+ if (handle.closed) {
+ return;
+ }
+ final long remaining = handle.usedMemoryInBytes;
+ if (remaining > 0L) {
+ memoryBlock.release(remaining);
+ handle.usedMemoryInBytes = 0L;
+ }
+ handles.remove(handle.id);
+ handle.closed = true;
+ }
+
+ private synchronized int activeHandleCountUnderLock() {
+ int count = 0;
+ for (final MemoryHandle handle : handles.values()) {
+ // An inactive queue keeps its share until lifecycle cleanup releases
its allocations.
+ if (!handle.closed && (handle.active || handle.usedMemoryInBytes > 0L)) {
+ count++;
+ }
+ }
+ return Math.max(1, count);
+ }
+
+ public enum AllocationRejectionReason {
+ NONE("NONE"),
+ MEMORY_QUOTA("SUBSCRIPTION_MEMORY_QUOTA"),
+ MEMORY_LIMIT("SUBSCRIPTION_MEMORY_LIMIT"),
+ OVERSIZED_ENTRY("SUBSCRIPTION_OVERSIZED_ENTRY");
+
+ private final String code;
+
+ AllocationRejectionReason(final String code) {
+ this.code = code;
+ }
+
+ public String getCode() {
+ return code;
+ }
+ }
+
+ public static final class AllocationResult {
+ private final boolean accepted;
+ private final AllocationRejectionReason rejectionReason;
+ private final long quotaInBytes;
+ private final long freeMemoryInBytes;
+
+ private AllocationResult(
+ final boolean accepted,
+ final AllocationRejectionReason rejectionReason,
+ final long quotaInBytes,
+ final long freeMemoryInBytes) {
+ this.accepted = accepted;
+ this.rejectionReason = rejectionReason;
+ this.quotaInBytes = quotaInBytes;
+ this.freeMemoryInBytes = freeMemoryInBytes;
+ }
+
+ private static AllocationResult accepted(
+ final long quotaInBytes, final long freeMemoryInBytes) {
+ return new AllocationResult(
+ true, AllocationRejectionReason.NONE, quotaInBytes,
freeMemoryInBytes);
+ }
+
+ private static AllocationResult rejected(
+ final AllocationRejectionReason reason,
+ final long quotaInBytes,
+ final long freeMemoryInBytes) {
+ return new AllocationResult(false, reason, quotaInBytes,
freeMemoryInBytes);
+ }
+
+ public boolean isAccepted() {
+ return accepted;
+ }
+
+ public AllocationRejectionReason getRejectionReason() {
+ return rejectionReason;
+ }
+
+ public long getQuotaInBytes() {
+ return quotaInBytes;
+ }
+
+ public long getFreeMemoryInBytes() {
+ return freeMemoryInBytes;
+ }
+ }
+
+ /** Per-queue view of the shared subscription memory budget. */
+ public final class MemoryHandle implements AutoCloseable {
+ private final long id;
+ private long usedMemoryInBytes;
+ private boolean active = true;
+ private boolean closed;
+
+ private MemoryHandle(final long id) {
+ this.id = id;
+ }
+
+ public AllocationResult tryAllocate(final long sizeInBytes) {
+ return SubscriptionMemoryManager.this.tryAllocate(this, sizeInBytes);
+ }
+
+ public void release(final long sizeInBytes) {
+ SubscriptionMemoryManager.this.release(this, sizeInBytes);
+ }
+
+ public long getUsedMemorySizeInBytes() {
+ synchronized (SubscriptionMemoryManager.this) {
+ return usedMemoryInBytes;
+ }
+ }
+
+ public long getMemoryQuotaInBytes() {
+ synchronized (SubscriptionMemoryManager.this) {
+ return getMemoryQuotaInBytesUnderLock();
+ }
+ }
+
+ private long getMemoryQuotaInBytesUnderLock() {
+ return Math.max(0L, getTotalMemorySizeInBytes() -
legacyUsedMemoryInBytes)
+ / activeHandleCountUnderLock();
+ }
+
+ public long getMaximumMemorySizeInBytes() {
+ synchronized (SubscriptionMemoryManager.this) {
+ return getMaximumMemorySizeInBytesUnderLock();
+ }
+ }
+
+ private long getMaximumMemorySizeInBytesUnderLock() {
+ final long budget = Math.max(0L, getTotalMemorySizeInBytes() -
legacyUsedMemoryInBytes);
+ final long protectedShare = getMemoryQuotaInBytesUnderLock() /
PROTECTED_SHARE_DIVISOR;
+ return Math.max(0L, budget - protectedShare *
(activeHandleCountUnderLock() - 1L));
+ }
+
+ public long getFreeMemorySizeInBytes() {
+ synchronized (SubscriptionMemoryManager.this) {
+ // Probe the size of one byte to account for protected shares and live
borrowed allocations.
+ return Math.max(
+ 0L,
+ SubscriptionMemoryManager.this
+ .inspectAllocationUnderLock(this, 1L)
+ .getFreeMemoryInBytes());
+ }
+ }
+
+ public boolean isActive() {
+ synchronized (SubscriptionMemoryManager.this) {
+ return active && !closed;
+ }
+ }
+
+ public void setActive(final boolean active) {
+ synchronized (SubscriptionMemoryManager.this) {
+ if (!closed) {
+ this.active = active;
+ }
+ }
+ }
+
+ public boolean canAllocate(final long sizeInBytes) {
+ synchronized (SubscriptionMemoryManager.this) {
+ return SubscriptionMemoryManager.this
+ .inspectAllocationUnderLock(this, sizeInBytes)
+ .isAccepted();
+ }
+ }
+
+ public AllocationRejectionReason inspectRejection(final long sizeInBytes) {
+ synchronized (SubscriptionMemoryManager.this) {
+ return SubscriptionMemoryManager.this
+ .inspectAllocationUnderLock(this, sizeInBytes)
+ .getRejectionReason();
+ }
+ }
+
+ @Override
+ public void close() {
+ SubscriptionMemoryManager.this.unregister(this);
+ }
+ }
}
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..c7ebf06ab05 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
@@ -26,12 +26,15 @@ import
org.apache.iotdb.consensus.iot.IoTConsensusServerImpl;
import org.apache.iotdb.consensus.iot.SubscriptionWalRetentionPolicy;
import org.apache.iotdb.consensus.iot.WriterSafeFrontierTracker;
import org.apache.iotdb.consensus.iot.log.ConsensusReqReader;
+import org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueAdmission;
+import
org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueRejectionReason;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.pipe.resource.memory.PipeMemoryWeightUtil;
import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils;
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.ErrorPayload;
import org.apache.iotdb.rpc.subscription.payload.poll.RegionProgress;
import org.apache.tsfile.enums.ColumnCategory;
@@ -65,6 +68,192 @@ public class ConsensusPrefetchingQueueDataNodeMemoryTest {
@Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+ @Test
+ public void
testLargeQueueQuotaDoesNotBlockSmallQueueAndPreservesPendingEntries()
+ throws Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final int originalBatchMaxDelay =
+
CommonDescriptor.getInstance().getConfig().getSubscriptionConsensusBatchMaxDelayInMs();
+ final int originalMaxWalEntries =
+
CommonDescriptor.getInstance().getConfig().getSubscriptionConsensusBatchMaxWalEntries();
+ ConsensusPrefetchingQueue largeQueue = null;
+ ConsensusPrefetchingQueue smallQueue = null;
+ try {
+
CommonDescriptor.getInstance().getConfig().setSubscriptionConsensusBatchMaxDelayInMs(0);
+
CommonDescriptor.getInstance().getConfig().setSubscriptionConsensusBatchMaxWalEntries(1);
+ final Tablet largeTablet = createLargeTablet();
+ final long largeBytes =
PipeMemoryWeightUtil.calculateTabletSizeInBytes(largeTablet);
+ final long smallBytes =
PipeMemoryWeightUtil.calculateTabletSizeInBytes(createTablet());
+ final SubscriptionMemoryManager memoryManager =
+ // With two queues, a queue may use 3/4 of the budget; make that
exactly one large entry.
+ new SubscriptionMemoryManager(largeBytes + largeBytes / 3L);
+ final ConsensusSubscriptionCommitManager commitManager =
+
newCommitManager(temporaryFolder.newFolder("large-small-fair-share"));
+ final FakeConsensusReqReader largeReader = new FakeConsensusReqReader();
+ final FakeConsensusReqReader smallReader = new FakeConsensusReqReader();
+ final AtomicInteger largeConversions = new AtomicInteger();
+ final AtomicInteger smallConversions = new AtomicInteger();
+ largeQueue =
+ newQueue(
+ "largeGroup",
+ new DataRegionId(1),
+ largeReader,
+ newConverter(largeConversions, largeTablet),
+ commitManager,
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE);
+ smallQueue =
+ newQueue(
+ "smallGroup",
+ new DataRegionId(2),
+ smallReader,
+ newConverter(smallConversions),
+ commitManager,
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE);
+ largeQueue.setSubscriptionMemoryManager(memoryManager);
+ smallQueue.setSubscriptionMemoryManager(memoryManager);
+ assertNull(largeQueue.poll("largeConsumer"));
+ assertNull(smallQueue.poll("smallConsumer"));
+
+ largeReader.currentSearchIndex = 2L;
+ assertTrue(pendingEntries(largeQueue).offer(createRequest(1L)));
+ assertTrue(pendingEntries(largeQueue).offer(createRequest(2L)));
+ largeQueue.drivePrefetchOnce();
+ assertEquals(largeBytes, largeQueue.getRetainedTabletBytes());
+ final long retainedPendingBytes = largeQueue.getRetainedRequestBytes();
+ assertTrue(retainedPendingBytes > 0L);
+ largeQueue.drivePrefetchOnce();
+ assertEquals(1, largeConversions.get());
+ assertEquals(1, pendingEntries(largeQueue).size());
+ assertEquals(retainedPendingBytes, largeQueue.getRetainedRequestBytes());
+ assertFalse(pendingEntries(largeQueue).offer(createRequest(3L)));
+ assertEquals(
+ SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_QUOTA,
+ ((SubscriptionQueueAdmission)
pendingEntries(largeQueue)).getLastRejectionReason());
+ assertEquals(1L, largeQueue.getRealtimeAdmissionRejectionCount());
+ assertEquals(
+ 1L,
+ largeQueue.getRealtimeAdmissionRejectionCount(
+ SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_QUOTA));
+
+ smallReader.currentSearchIndex = 1L;
+ assertTrue(pendingEntries(smallQueue).offer(createRequest(1L)));
+ smallQueue.drivePrefetchOnce();
+ assertEquals(1, smallConversions.get());
+ assertEquals(largeBytes + smallBytes,
memoryManager.getUsedMemorySizeInBytes());
+ final SubscriptionEvent smallEvent = smallQueue.poll("smallConsumer");
+ assertNotNull(smallEvent);
+ assertTrue(smallQueue.ack("smallConsumer",
smallEvent.getCommitContext()));
+ assertEquals(largeBytes, memoryManager.getUsedMemorySizeInBytes());
+
+ final SubscriptionEvent largeEvent = largeQueue.poll("largeConsumer");
+ assertNotNull(largeEvent);
+ assertTrue(largeQueue.ack("largeConsumer",
largeEvent.getCommitContext()));
+ largeQueue.drivePrefetchOnce();
+ assertEquals(2, largeConversions.get());
+ assertEquals(0, pendingEntries(largeQueue).size());
+ assertEquals(0L, largeQueue.getRetainedRequestBytes());
+ final SubscriptionEvent secondLargeEvent =
largeQueue.poll("largeConsumer");
+ assertNotNull(secondLargeEvent);
+ assertTrue(largeQueue.ack("largeConsumer",
secondLargeEvent.getCommitContext()));
+ assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
+ } finally {
+ if (largeQueue != null) {
+ largeQueue.close();
+ }
+ if (smallQueue != null) {
+ smallQueue.close();
+ }
+ CommonDescriptor.getInstance()
+ .getConfig()
+ .setSubscriptionConsensusBatchMaxDelayInMs(originalBatchMaxDelay);
+ CommonDescriptor.getInstance()
+ .getConfig()
+ .setSubscriptionConsensusBatchMaxWalEntries(originalMaxWalEntries);
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
+ @Test
+ public void testOversizedEntryReportsErrorWhileOtherQueueHoldsMemory()
throws Exception {
+ final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
+ final int originalBatchMaxDelay =
+
CommonDescriptor.getInstance().getConfig().getSubscriptionConsensusBatchMaxDelayInMs();
+ ConsensusPrefetchingQueue largeQueue = null;
+ ConsensusPrefetchingQueue smallQueue = null;
+ try {
+
CommonDescriptor.getInstance().getConfig().setSubscriptionConsensusBatchMaxDelayInMs(0);
+ final Tablet largeTablet = createLargeTablet();
+ final long largeBytes =
PipeMemoryWeightUtil.calculateTabletSizeInBytes(largeTablet);
+ // This entry exceeds the entire node budget, which is also partly held
by another queue.
+ final SubscriptionMemoryManager memoryManager =
+ new SubscriptionMemoryManager(largeBytes / 2L);
+ final ConsensusSubscriptionCommitManager commitManager =
+ newCommitManager(temporaryFolder.newFolder("oversized-queue-error"));
+ final FakeConsensusReqReader largeReader = new FakeConsensusReqReader();
+ final FakeConsensusReqReader smallReader = new FakeConsensusReqReader();
+ final AtomicInteger largeConversions = new AtomicInteger();
+ largeQueue =
+ newQueue(
+ "largeGroup",
+ new DataRegionId(1),
+ largeReader,
+ newConverter(largeConversions, largeTablet),
+ commitManager,
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE);
+ smallQueue =
+ newQueue(
+ "smallGroup",
+ new DataRegionId(2),
+ smallReader,
+ newConverter(new AtomicInteger()),
+ commitManager,
+ TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE);
+ largeQueue.setSubscriptionMemoryManager(memoryManager);
+ smallQueue.setSubscriptionMemoryManager(memoryManager);
+ assertNull(largeQueue.poll("largeConsumer"));
+ assertNull(smallQueue.poll("smallConsumer"));
+ smallReader.currentSearchIndex = 1L;
+ assertTrue(pendingEntries(smallQueue).offer(createRequest(1L)));
+ smallQueue.drivePrefetchOnce();
+ final long smallBytes = memoryManager.getUsedMemorySizeInBytes();
+ assertTrue(smallBytes > 0L);
+
+ largeReader.currentSearchIndex = 1L;
+ assertTrue(pendingEntries(largeQueue).offer(createRequest(1L)));
+ largeQueue.drivePrefetchOnce();
+ assertEquals(1, largeConversions.get());
+ assertEquals(1L, largeQueue.getCurrentReadSearchIndex());
+ assertEquals(1L, largeQueue.getOversizedEntryRejectionCount());
+ assertEquals(0L, largeQueue.getRealtimeAdmissionRejectionCount());
+ assertEquals(smallBytes, memoryManager.getUsedMemorySizeInBytes());
+ final SubscriptionEvent error = largeQueue.poll("largeConsumer");
+ assertNotNull(error);
+ final ErrorPayload payload = (ErrorPayload)
error.getCurrentResponse().getPayload();
+ assertTrue(payload.isCritical());
+
assertTrue(payload.getErrorMessage().contains("SUBSCRIPTION_OVERSIZED_ENTRY"));
+ for (int i = 0; i < 10; i++) {
+ largeQueue.drivePrefetchOnce();
+ }
+ assertEquals(1, largeConversions.get());
+ assertEquals(1L, largeQueue.getCurrentReadSearchIndex());
+ final SubscriptionEvent smallEvent = smallQueue.poll("smallConsumer");
+ assertNotNull(smallEvent);
+ assertTrue(smallQueue.ack("smallConsumer",
smallEvent.getCommitContext()));
+ assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
+ } finally {
+ if (largeQueue != null) {
+ largeQueue.close();
+ }
+ if (smallQueue != null) {
+ smallQueue.close();
+ }
+ CommonDescriptor.getInstance()
+ .getConfig()
+ .setSubscriptionConsensusBatchMaxDelayInMs(originalBatchMaxDelay);
+
IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir);
+ }
+ }
+
@Test
public void testQueuesShareDataNodeMemoryBudget() throws Exception {
final String originalSystemDir =
IoTDBDescriptor.getInstance().getConfig().getSystemDir();
@@ -77,7 +266,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 * 2L);
final ConsensusSubscriptionCommitManager commitManager =
newCommitManager(systemDir);
final FakeConsensusReqReader readerA = new FakeConsensusReqReader();
@@ -121,34 +311,28 @@ public class ConsensusPrefetchingQueueDataNodeMemoryTest {
<= memoryManager.getTotalMemorySizeInBytes());
readerB.currentSearchIndex = 1L;
- assertFalse(pendingEntries(queueB).offer(createRequest(1L)));
+ assertTrue(pendingEntries(queueB).offer(createRequest(1L)));
queueB.drivePrefetchOnce();
- assertEquals(0, conversionCountB.get());
- assertEquals("true",
queueB.coreReportMessage().get("realtimeAdmissionBlocked"));
- assertEquals(oneTabletBytes, memoryManager.getUsedMemorySizeInBytes());
+ assertEquals(1, conversionCountB.get());
+ assertEquals("false",
queueB.coreReportMessage().get("realtimeAdmissionBlocked"));
+ assertEquals(oneTabletBytes * 2L,
memoryManager.getUsedMemorySizeInBytes());
+ assertEquals(oneTabletBytes, queueB.getRetainedTabletBytes());
final SubscriptionEvent eventA = queueA.poll("consumerA");
assertNotNull(eventA);
assertTrue(queueA.ack("consumerA", eventA.getCommitContext()));
assertEquals(0L, queueA.getRetainedTabletBytes());
- assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
+ assertEquals(oneTabletBytes, memoryManager.getUsedMemorySizeInBytes());
queueB.drivePrefetchOnce();
assertEquals("false",
queueB.coreReportMessage().get("realtimeAdmissionBlocked"));
- assertTrue(pendingEntries(queueB).offer(createRequest(1L)));
- queueB.drivePrefetchOnce();
-
- assertEquals(1, conversionCountB.get());
- assertEquals(oneTabletBytes, queueB.getRetainedTabletBytes());
- assertEquals(oneTabletBytes, memoryManager.getUsedMemorySizeInBytes());
- assertTrue(
- queueA.getRetainedTabletBytes() + queueB.getRetainedTabletBytes()
- <= memoryManager.getTotalMemorySizeInBytes());
final SubscriptionEvent eventB = queueB.poll("consumerB");
assertNotNull(eventB);
assertTrue(queueB.ack("consumerB", eventB.getCommitContext()));
assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
+ queueB.drivePrefetchOnce();
+ assertEquals("false",
queueB.coreReportMessage().get("realtimeAdmissionBlocked"));
} finally {
if (queueA != null) {
queueA.close();
@@ -203,23 +387,20 @@ public class ConsensusPrefetchingQueueDataNodeMemoryTest {
assertEquals(1, conversionCount.get());
assertEquals(2L, queue.getCurrentReadSearchIndex());
assertEquals("1",
queue.coreReportMessage().get("bufferedRealtimeEntryCount"));
- assertEquals("0", queue.coreReportMessage().get("pendingEntriesSize"));
+ assertEquals("1", queue.coreReportMessage().get("pendingEntriesSize"));
assertEquals("true",
queue.coreReportMessage().get("realtimeAdmissionBlocked"));
+ assertFalse(pendingEntries(queue).offer(createRequest(3L)));
+ assertEquals("SUBSCRIPTION_WRITER_BACKLOG",
queue.getLastAdmissionRejectionCode());
+ assertEquals(0L, queue.getSubscriptionMemoryRejectionCount());
queue.setActiveWriterNodeIds(Collections.singleton(7));
queue.drivePrefetchOnce();
- assertEquals(1, conversionCount.get());
- assertEquals("0",
queue.coreReportMessage().get("bufferedRealtimeEntryCount"));
- assertEquals("false",
queue.coreReportMessage().get("realtimeAdmissionBlocked"));
-
- assertTrue(pendingEntries(queue).offer(createRequest(2L)));
- queue.drivePrefetchOnce();
-
assertEquals(2, conversionCount.get());
assertEquals(3L, queue.getCurrentReadSearchIndex());
assertEquals("0",
queue.coreReportMessage().get("bufferedRealtimeEntryCount"));
assertEquals("0", queue.coreReportMessage().get("pendingEntriesSize"));
+ assertEquals("false",
queue.coreReportMessage().get("realtimeAdmissionBlocked"));
} finally {
if (queue != null) {
queue.close();
@@ -260,12 +441,17 @@ public class ConsensusPrefetchingQueueDataNodeMemoryTest {
}
private static ConsensusLogToTabletConverter newConverter(final
AtomicInteger conversionCount) {
+ return newConverter(conversionCount, createTablet());
+ }
+
+ private static ConsensusLogToTabletConverter newConverter(
+ final AtomicInteger conversionCount, final Tablet tablet) {
final ConsensusLogToTabletConverter converter =
mock(ConsensusLogToTabletConverter.class);
when(converter.convert(any()))
.thenAnswer(
ignored -> {
conversionCount.incrementAndGet();
- return Collections.singletonList(createTablet());
+ return Collections.singletonList(tablet);
});
when(converter.getDatabaseName()).thenReturn("db");
return converter;
@@ -291,6 +477,21 @@ public class ConsensusPrefetchingQueueDataNodeMemoryTest {
return tablet;
}
+ private static Tablet createLargeTablet() {
+ final Tablet tablet =
+ new Tablet(
+ "sensors",
+ Arrays.asList("device", "payload"),
+ Arrays.asList(TSDataType.STRING, TSDataType.STRING),
+ Arrays.asList(ColumnCategory.TAG, ColumnCategory.FIELD),
+ 1);
+ tablet.addTimestamp(0, 1L);
+ tablet.addValue(0, 0, "d1");
+ tablet.addValue(0, 1, "v".repeat(20_000));
+ tablet.setRowSize(1);
+ return tablet;
+ }
+
private static IndexedConsensusRequest createRequest(final long searchIndex)
{
return new IndexedConsensusRequest(
searchIndex,
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 190a21cca3f..2b3dc9f0ef5 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
@@ -30,6 +30,8 @@ import
org.apache.iotdb.consensus.iot.SubscriptionWalRetentionPolicy;
import org.apache.iotdb.consensus.iot.WriterSafeFrontierTracker;
import org.apache.iotdb.consensus.iot.log.ConsensusReqReader;
import org.apache.iotdb.consensus.iot.logdispatcher.IoTConsensusMemoryManager;
+import org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueAdmission;
+import
org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueRejectionReason;
import org.apache.iotdb.db.conf.IoTDBDescriptor;
import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils;
import org.apache.iotdb.db.storageengine.dataregion.wal.io.ProgressWALReader;
@@ -81,8 +83,9 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantReadWriteLock;
-import java.util.function.BooleanSupplier;
+import java.util.function.Consumer;
import java.util.function.LongSupplier;
+import java.util.function.Supplier;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -636,7 +639,8 @@ public class ConsensusPrefetchingQueueTest {
final Class<?> queueClass =
Class.forName(ConsensusPrefetchingQueue.class.getName() +
"$WakeableIndexedConsensusQueue");
final Constructor<?> constructor =
- queueClass.getDeclaredConstructor(int.class, Runnable.class,
BooleanSupplier.class);
+ queueClass.getDeclaredConstructor(
+ int.class, Runnable.class, Supplier.class, Consumer.class);
constructor.setAccessible(true);
final AtomicBoolean admissionEnabled = new AtomicBoolean(true);
@@ -646,7 +650,7 @@ public class ConsensusPrefetchingQueueTest {
final CountDownLatch clearCompleted = new CountDownLatch(1);
final AtomicReference<Boolean> offered = new AtomicReference<>();
final AtomicReference<Throwable> asyncFailure = new AtomicReference<>();
- final BooleanSupplier admissionSupplier =
+ final Supplier<SubscriptionQueueRejectionReason> admissionSupplier =
() -> {
final boolean admitted = admissionEnabled.get();
admissionChecked.countDown();
@@ -658,11 +662,17 @@ public class ConsensusPrefetchingQueueTest {
Thread.currentThread().interrupt();
throw new AssertionError(e);
}
- return admitted;
+ return admitted
+ ? SubscriptionQueueRejectionReason.NONE
+ : SubscriptionQueueRejectionReason.INACTIVE_OR_CLOSED;
};
final BlockingQueue<IndexedConsensusRequest> queue =
(BlockingQueue<IndexedConsensusRequest>)
- constructor.newInstance(8, (Runnable) () -> {}, admissionSupplier);
+ constructor.newInstance(
+ 8,
+ (Runnable) () -> {},
+ admissionSupplier,
+ (Consumer<SubscriptionQueueRejectionReason>) reason -> {});
final Thread offerThread =
new Thread(
@@ -715,6 +725,77 @@ public class ConsensusPrefetchingQueueTest {
}
assertTrue(Boolean.TRUE.equals(offered.get()));
assertTrue(queue.isEmpty());
+ assertFalse(queue.offer(createRequest(2L)));
+ assertEquals(
+ SubscriptionQueueRejectionReason.INACTIVE_OR_CLOSED,
+ ((SubscriptionQueueAdmission) queue).getLastRejectionReason());
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testConcurrentOffersKeepTheirOwnRejectionReasons() throws
Exception {
+ final Class<?> queueClass =
+ Class.forName(ConsensusPrefetchingQueue.class.getName() +
"$WakeableIndexedConsensusQueue");
+ final Constructor<?> constructor =
+ queueClass.getDeclaredConstructor(
+ int.class, Runnable.class, Supplier.class, Consumer.class);
+ constructor.setAccessible(true);
+ final BlockingQueue<IndexedConsensusRequest> queue =
+ (BlockingQueue<IndexedConsensusRequest>)
+ constructor.newInstance(
+ 8,
+ (Runnable) () -> {},
+ (Supplier<SubscriptionQueueRejectionReason>)
+ () ->
+ Thread.currentThread().getName().equals("first-offer")
+ ?
SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_QUOTA
+ : SubscriptionQueueRejectionReason.WRITER_BACKLOG,
+ (Consumer<SubscriptionQueueRejectionReason>) reason -> {});
+ final CountDownLatch firstOffered = new CountDownLatch(1);
+ final CountDownLatch secondOffered = new CountDownLatch(1);
+ final AtomicReference<SubscriptionQueueRejectionReason> firstReason = new
AtomicReference<>();
+ final AtomicReference<SubscriptionQueueRejectionReason> secondReason = new
AtomicReference<>();
+ final AtomicReference<Throwable> asyncFailure = new AtomicReference<>();
+ final Thread first =
+ new Thread(
+ () -> {
+ try {
+ assertFalse(queue.offer(createRequest(1L)));
+ firstOffered.countDown();
+ assertTrue(secondOffered.await(5, TimeUnit.SECONDS));
+ firstReason.set(((SubscriptionQueueAdmission)
queue).getLastRejectionReason());
+ } catch (final Throwable t) {
+ asyncFailure.compareAndSet(null, t);
+ }
+ },
+ "first-offer");
+ final Thread second =
+ new Thread(
+ () -> {
+ try {
+ assertTrue(firstOffered.await(5, TimeUnit.SECONDS));
+ assertFalse(queue.offer(createRequest(2L)));
+ secondReason.set(((SubscriptionQueueAdmission)
queue).getLastRejectionReason());
+ } catch (final Throwable t) {
+ asyncFailure.compareAndSet(null, t);
+ } finally {
+ secondOffered.countDown();
+ }
+ },
+ "second-offer");
+ first.setDaemon(true);
+ second.setDaemon(true);
+ first.start();
+ second.start();
+ first.join(TimeUnit.SECONDS.toMillis(5));
+ second.join(TimeUnit.SECONDS.toMillis(5));
+ assertFalse(first.isAlive());
+ assertFalse(second.isAlive());
+ if (asyncFailure.get() != null) {
+ throw new AssertionError(asyncFailure.get());
+ }
+ assertEquals(SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_QUOTA,
firstReason.get());
+ assertEquals(SubscriptionQueueRejectionReason.WRITER_BACKLOG,
secondReason.get());
}
@Test
@@ -723,7 +804,8 @@ public class ConsensusPrefetchingQueueTest {
final Class<?> queueClass =
Class.forName(ConsensusPrefetchingQueue.class.getName() +
"$WakeableIndexedConsensusQueue");
final Constructor<?> constructor =
- queueClass.getDeclaredConstructor(int.class, Runnable.class,
BooleanSupplier.class);
+ queueClass.getDeclaredConstructor(
+ int.class, Runnable.class, Supplier.class, Consumer.class);
constructor.setAccessible(true);
final Method retainedBytesMethod =
queueClass.getDeclaredMethod("getRetainedRequestBytes");
retainedBytesMethod.setAccessible(true);
@@ -735,10 +817,20 @@ public class ConsensusPrefetchingQueueTest {
new AtomicLongMemoryBlock("SubscriptionPendingTest", null, 100L);
final BlockingQueue<IndexedConsensusRequest> firstQueue =
(BlockingQueue<IndexedConsensusRequest>)
- constructor.newInstance(8, (Runnable) () -> {}, (BooleanSupplier)
() -> true);
+ constructor.newInstance(
+ 8,
+ (Runnable) () -> {},
+ (Supplier<SubscriptionQueueRejectionReason>)
+ () -> SubscriptionQueueRejectionReason.NONE,
+ (Consumer<SubscriptionQueueRejectionReason>) reason -> {});
final BlockingQueue<IndexedConsensusRequest> secondQueue =
(BlockingQueue<IndexedConsensusRequest>)
- constructor.newInstance(8, (Runnable) () -> {}, (BooleanSupplier)
() -> true);
+ constructor.newInstance(
+ 8,
+ (Runnable) () -> {},
+ (Supplier<SubscriptionQueueRejectionReason>)
+ () -> SubscriptionQueueRejectionReason.NONE,
+ (Consumer<SubscriptionQueueRejectionReason>) reason -> {});
memoryManager.init(testMemoryBlock, 0.6);
try {
@@ -750,6 +842,9 @@ public class ConsensusPrefetchingQueueTest {
assertEquals(40L, retainedBytesMethod.invoke(secondQueue));
assertFalse(firstQueue.offer(createSizedRequest(2L, 10L, 30)));
+ assertEquals(
+ SubscriptionQueueRejectionReason.CONSENSUS_REQUEST_MEMORY_LIMIT,
+ ((SubscriptionQueueAdmission) firstQueue).getLastRejectionReason());
assertEquals(40L, testMemoryBlock.getUsedMemoryInBytes());
firstQueue.clear();
@@ -989,6 +1084,7 @@ public class ConsensusPrefetchingQueueTest {
1L,
1L,
true);
+ queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(1024L *
1024L));
final IndexedConsensusRequest dataRequest =
new IndexedConsensusRequest(
@@ -1809,6 +1905,7 @@ public class ConsensusPrefetchingQueueTest {
1L,
1L,
true);
+ queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(1024L *
1024L));
final IndexedConsensusRequest dataRequest =
new IndexedConsensusRequest(
@@ -1877,6 +1974,7 @@ public class ConsensusPrefetchingQueueTest {
1L,
1L,
true);
+ queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(1024L *
1024L));
final IndexedConsensusRequest dataRequest =
new IndexedConsensusRequest(
@@ -2685,6 +2783,7 @@ public class ConsensusPrefetchingQueueTest {
1L,
1L,
true);
+ queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(1024L *
1024L));
reader.currentSearchIndex = 2L;
assertTrue(pendingEntries(queue).offer(createRequest(1L)));
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 b3783beab43..e25c77803cf 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
@@ -154,7 +154,7 @@ public class ConsensusPrefetchingQueueWalBackpressureTest {
assertEquals(Arrays.asList(1L, 2L), conversionAttempts);
assertEquals(2L, queue.getCurrentReadSearchIndex());
assertEquals(1, queue.getPrefetchedEventCount());
- assertTrue(pendingEntries(queue).isEmpty());
+ assertEquals(0, pendingEntries(queue).size());
assertEquals("true",
queue.coreReportMessage().get("realtimeAdmissionBlocked"));
assertEquals(oneTabletBytes, memoryManager.getUsedMemorySizeInBytes());
assertMemoryBounded(queue, memoryManager);
@@ -307,22 +307,11 @@ public class ConsensusPrefetchingQueueWalBackpressureTest
{
}
@Override
- public synchronized boolean tryAllocate(final long sizeInBytes) {
- final boolean allocated = super.tryAllocate(sizeInBytes);
- observeUsage();
- return allocated;
- }
-
- @Override
- public synchronized void release(final long sizeInBytes) {
- super.release(sizeInBytes);
- observeUsage();
- }
-
- private void observeUsage() {
- final long usedMemorySizeInBytes = getUsedMemorySizeInBytes();
+ public synchronized long getUsedMemorySizeInBytes() {
+ final long usedMemorySizeInBytes = super.getUsedMemorySizeInBytes();
maximumUsedMemorySizeInBytes = Math.max(maximumUsedMemorySizeInBytes,
usedMemorySizeInBytes);
exceededBudget |= usedMemorySizeInBytes > getTotalMemorySizeInBytes();
+ return usedMemorySizeInBytes;
}
private long getMaximumUsedMemorySizeInBytes() {
diff --git
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetricsTest.java
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetricsTest.java
index 08880f4fcb3..782cebe30cd 100644
---
a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetricsTest.java
+++
b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/metric/ConsensusSubscriptionPrefetchingQueueMetricsTest.java
@@ -22,6 +22,7 @@ package org.apache.iotdb.db.subscription.metric;
import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.service.metric.enums.Metric;
import org.apache.iotdb.commons.service.metric.enums.Tag;
+import
org.apache.iotdb.consensus.iot.subscription.SubscriptionQueueRejectionReason;
import
org.apache.iotdb.db.subscription.broker.consensus.ConsensusPrefetchingQueue;
import org.apache.iotdb.metrics.AbstractMetricService;
import org.apache.iotdb.metrics.type.Rate;
@@ -132,6 +133,18 @@ public class
ConsensusSubscriptionPrefetchingQueueMetricsTest {
eq(queueId),
eq(Tag.REGION.toString()),
eq(firstRegionId.toString()));
+ verify(metricService)
+ .createAutoGauge(
+
eq(Metric.SUBSCRIPTION_CONSENSUS_ADMISSION_REJECTIONS_BY_REASON.toString()),
+ eq(MetricLevel.IMPORTANT),
+ eq(firstQueue),
+ any(),
+ eq(Tag.NAME.toString()),
+ eq(queueId),
+ eq(Tag.REGION.toString()),
+ eq(firstRegionId.toString()),
+ eq(Tag.REASON.toString()),
+
eq(SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_QUOTA.getCode()));
verify(metricService)
.createAutoGauge(
eq(Metric.SUBSCRIPTION_CONSENSUS_LAG.toString()),
@@ -164,6 +177,16 @@ public class
ConsensusSubscriptionPrefetchingQueueMetricsTest {
queueId,
Tag.REGION.toString(),
firstRegionId.toString());
+ verify(metricService)
+ .remove(
+ MetricType.AUTO_GAUGE,
+
Metric.SUBSCRIPTION_CONSENSUS_ADMISSION_REJECTIONS_BY_REASON.toString(),
+ Tag.NAME.toString(),
+ queueId,
+ Tag.REGION.toString(),
+ firstRegionId.toString(),
+ Tag.REASON.toString(),
+
SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_QUOTA.getCode());
verify(metricService, never())
.remove(
MetricType.AUTO_GAUGE,
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..beda24e623b 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
@@ -61,4 +61,112 @@ public class SubscriptionMemoryManagerTest {
assertFalse(memoryManager.tryAllocate(1L));
assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
}
+
+ @Test
+ public void
testQueueCanBorrowIdleMemoryWithoutConsumingAnotherQueuesProtectedShare() {
+ final SubscriptionMemoryManager memoryManager = new
SubscriptionMemoryManager(10L);
+ final SubscriptionMemoryManager.MemoryHandle queueA =
memoryManager.registerQueue();
+ final SubscriptionMemoryManager.MemoryHandle queueB =
memoryManager.registerQueue();
+
+ assertEquals(5L, queueA.getMemoryQuotaInBytes());
+ assertEquals(
+ SubscriptionMemoryManager.AllocationRejectionReason.OVERSIZED_ENTRY,
+ queueA.inspectRejection(9L));
+ assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
+
+ assertTrue(queueA.tryAllocate(6L).isAccepted());
+ assertTrue(queueA.tryAllocate(2L).isAccepted());
+ assertEquals(
+ SubscriptionMemoryManager.AllocationRejectionReason.MEMORY_QUOTA,
+ queueA.inspectRejection(1L));
+ assertTrue(queueB.tryAllocate(2L).isAccepted());
+ assertFalse(queueB.tryAllocate(1L).isAccepted());
+ assertEquals(10L, memoryManager.getUsedMemorySizeInBytes());
+
+ queueA.close();
+ assertEquals(10L, queueB.getMemoryQuotaInBytes());
+ assertTrue(queueB.tryAllocate(8L).isAccepted());
+ assertEquals(10L, memoryManager.getUsedMemorySizeInBytes());
+
+ queueB.close();
+ assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
+ }
+
+ @Test
+ public void
testBorrowedMemoryWaitsForAckWhenAnotherQueueNeedsMoreThanItsReserve() {
+ final SubscriptionMemoryManager memoryManager = new
SubscriptionMemoryManager(100L);
+ try (SubscriptionMemoryManager.MemoryHandle largeQueue =
memoryManager.registerQueue();
+ SubscriptionMemoryManager.MemoryHandle smallQueue =
memoryManager.registerQueue()) {
+ assertTrue(largeQueue.tryAllocate(75L).isAccepted());
+ assertEquals(25L, smallQueue.getFreeMemorySizeInBytes());
+ assertEquals(
+ SubscriptionMemoryManager.AllocationRejectionReason.MEMORY_LIMIT,
+ smallQueue.inspectRejection(30L));
+ assertTrue(smallQueue.tryAllocate(25L).isAccepted());
+ assertEquals(0L, smallQueue.getFreeMemorySizeInBytes());
+ largeQueue.release(50L);
+ assertTrue(smallQueue.tryAllocate(30L).isAccepted());
+ assertEquals(80L, memoryManager.getUsedMemorySizeInBytes());
+ }
+ }
+
+ @Test
+ public void testNewQueueWaitsForExistingBorrowedAllocationsToDrain() {
+ final SubscriptionMemoryManager memoryManager = new
SubscriptionMemoryManager(100L);
+ try (SubscriptionMemoryManager.MemoryHandle first =
memoryManager.registerQueue()) {
+ assertTrue(first.tryAllocate(100L).isAccepted());
+ try (SubscriptionMemoryManager.MemoryHandle second =
memoryManager.registerQueue()) {
+ assertEquals(
+ SubscriptionMemoryManager.AllocationRejectionReason.MEMORY_LIMIT,
+ second.inspectRejection(1L));
+ first.release(25L);
+ assertTrue(second.tryAllocate(25L).isAccepted());
+ }
+ }
+ }
+
+ @Test
+ public void testOversizedQueueEntryIsExplicitlyRejectedEvenWhenNodeIsEmpty()
{
+ final SubscriptionMemoryManager memoryManager = new
SubscriptionMemoryManager(10L);
+ try (SubscriptionMemoryManager.MemoryHandle queue =
memoryManager.registerQueue()) {
+ assertEquals(
+ SubscriptionMemoryManager.AllocationRejectionReason.OVERSIZED_ENTRY,
+ queue.tryAllocate(11L).getRejectionReason());
+ assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
+ assertTrue(queue.tryAllocate(10L).isAccepted());
+ }
+ assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
+ }
+
+ @Test
+ public void testInactiveQueueKeepsShareUntilRetainedDataIsReleased() {
+ final SubscriptionMemoryManager memoryManager = new
SubscriptionMemoryManager(10L);
+ try (SubscriptionMemoryManager.MemoryHandle queueA =
memoryManager.registerQueue();
+ SubscriptionMemoryManager.MemoryHandle queueB =
memoryManager.registerQueue()) {
+ assertTrue(queueA.tryAllocate(5L).isAccepted());
+ queueA.setActive(false);
+ assertEquals(5L, queueB.getMemoryQuotaInBytes());
+ assertTrue(queueB.tryAllocate(5L).isAccepted());
+ queueA.release(5L);
+ assertEquals(10L, queueB.getMemoryQuotaInBytes());
+ assertTrue(queueB.tryAllocate(5L).isAccepted());
+ }
+ assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
+ }
+
+ @Test
+ public void testAllocationProbeDoesNotReserveMemory() {
+ final SubscriptionMemoryManager memoryManager = new
SubscriptionMemoryManager(10L);
+ try (SubscriptionMemoryManager.MemoryHandle queue =
memoryManager.registerQueue()) {
+ for (int i = 0; i < 100; i++) {
+ assertTrue(queue.canAllocate(10L));
+ assertEquals(
+ SubscriptionMemoryManager.AllocationRejectionReason.NONE,
queue.inspectRejection(10L));
+ }
+ assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
+ assertTrue(queue.tryAllocate(10L).isAccepted());
+ queue.release(100L);
+ assertEquals(0L, memoryManager.getUsedMemorySizeInBytes());
+ }
+ }
}
diff --git
a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
index 158a079be6b..b229f36d1fe 100644
---
a/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
+++
b/iotdb-core/node-commons/src/assembly/resources/conf/iotdb-system.properties.template
@@ -1206,6 +1206,13 @@ may_cache_nonexist_series=true
# The parameter form is a:b:c:d:e:f:g:h:i, where a, b, c, d, e, f, g, h and i
are integers. for example: 1:1:1:1:1:1:1:1:1 , 1:100:200:50:200:200:200:50:250
# The legacy eight-part form is still accepted and defaults Subscription to
20% of query memory.
# When subscription is disabled, its memory proportion is 0.
+# The ninth (Subscription) share controls consensus subscription
materialization, independently of
+# subscription_cache_memory_usage_percentage. Each active queue protects half
of its equal share;
+# the remaining idle capacity can be borrowed by another queue, but not
revoked until ACK. Size
+# the budget for the largest materialized WAL entry * active queue count * at
least 1.2; use a
+# larger multiplier for multiple buffered entries per queue. An entry
exceeding the maximum a
+# queue can hold under these protected shares reports
SUBSCRIPTION_OVERSIZED_ENTRY and does not
+# advance WAL progress. New queues may wait for existing borrowed allocations
to drain via ACK.
# effectiveMode: restart
chunk_timeseriesmeta_free_memory_proportion=1:100:200:50:200:200:200:50:250
@@ -2153,6 +2160,28 @@ pipe_all_sinks_rate_limit_bytes_per_second=-1
### Subscription Consensus Configuration
####################
+# Maximum fraction of pipe non-floating memory used to cache serialized
subscription poll responses.
+# Recommended starting value: 0.2, including large-field and large-batch
workloads. This does not control
+# consensus subscription tablet materialization; size that budget using the
ninth
+# chunk_timeseriesmeta_free_memory_proportion share instead. The cache may
receive less than this
+# maximum under pipe memory pressure.
+# effectiveMode: restart
+# Datatype: double, range: [0, 1]
+subscription_cache_memory_usage_percentage=0.2
+
+# Maximum number of consensus-subscription prefetch workers. For large-field
workloads, compare
+# 2 and 4 workers under load, checking CPU, heap and ACK latency; queues share
the worker pool.
+# effectiveMode: restart
+# Datatype: int
+subscription_consensus_prefetch_executor_max_thread_num=2
+
+# Maximum materialized consensus-subscription batch size. This is a batching
target, not a hard
+# limit on one WAL entry/tablet. For large-field workloads, choose a value at
least as large as a
+# typical single tablet if batching is desired, and provide enough
materialization memory per queue.
+# effectiveMode: restart
+# Datatype: long, Unit: byte
+subscription_consensus_batch_max_size_in_bytes=8388608
+
# The interval at which an idle consensus-subscription queue advances its
writer safe-time barrier by watermark.
# effectiveMode: restart
# Datatype: long(duration)
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..d412d74d0cc 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
@@ -232,6 +232,14 @@ public enum Metric {
SUBSCRIPTION_CONSENSUS_LAST_POLL_TIME("subscription_consensus_last_poll_time"),
SUBSCRIPTION_CONSENSUS_LAST_PROGRESS_TIME("subscription_consensus_last_progress_time"),
SUBSCRIPTION_CONSENSUS_PROGRESS_STATUS("subscription_consensus_progress_status"),
+ SUBSCRIPTION_CONSENSUS_MEMORY_USED("subscription_consensus_memory_used"),
+ SUBSCRIPTION_CONSENSUS_MEMORY_QUOTA("subscription_consensus_memory_quota"),
+
SUBSCRIPTION_CONSENSUS_MEMORY_MAXIMUM("subscription_consensus_memory_maximum"),
+
SUBSCRIPTION_CONSENSUS_ADMISSION_REJECTIONS("subscription_consensus_admission_rejections"),
+ SUBSCRIPTION_CONSENSUS_ADMISSION_REJECTIONS_BY_REASON(
+ "subscription_consensus_admission_rejections_by_reason"),
+
SUBSCRIPTION_CONSENSUS_MEMORY_REJECTIONS("subscription_consensus_memory_rejections"),
+
SUBSCRIPTION_CONSENSUS_OVERSIZED_REJECTIONS("subscription_consensus_oversized_rejections"),
// load related
ACTIVE_LOADING_FILES_NUMBER("active_loading_files_number"),
ACTIVE_LOADING_FILES_SIZE("active_loading_files_size"),
diff --git
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Tag.java
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Tag.java
index 91b4307e02d..0f0aa0dadc6 100644
---
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Tag.java
+++
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/service/metric/enums/Tag.java
@@ -23,6 +23,7 @@ public enum Tag {
TYPE("type"),
NAME("name"),
REGION("region"),
+ REASON("reason"),
STATUS("status"),
DATABASE("database"),
FROM("from"),
