This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch fix/subscription-fair-share-admission in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 57a1f842a7bf226bb07698bc1ddbf35a538c99c4 Author: Caideyipi <[email protected]> AuthorDate: Fri Oct 9 16:48:08 2026 +0800 Isolate subscription materialization memory by queue --- .../zh/UserGuide/Subscription/Memory-Management.md | 91 ++++++ .../iotdb/consensus/i18n/IoTConsensusMessages.java | 4 + .../iotdb/consensus/i18n/IoTConsensusMessages.java | 4 + .../subscription/SubscriptionQueueAdmission.java | 25 ++ .../subscription/SubscriptionQueueRegistry.java | 34 ++- .../SubscriptionQueueRejectionReason.java | 42 +++ .../apache/iotdb/db/i18n/DataNodePipeMessages.java | 2 + .../apache/iotdb/db/i18n/DataNodePipeMessages.java | 2 + .../consensus/ConsensusPrefetchingQueue.java | 240 +++++++++++++--- ...nsensusSubscriptionPrefetchingQueueMetrics.java | 48 ++++ .../resource/SubscriptionMemoryManager.java | 313 +++++++++++++++++++-- ...onsensusPrefetchingQueueDataNodeMemoryTest.java | 246 ++++++++++++++-- .../consensus/ConsensusPrefetchingQueueTest.java | 115 +++++++- ...nsensusPrefetchingQueueWalBackpressureTest.java | 19 +- .../resource/SubscriptionMemoryManagerTest.java | 75 +++++ .../conf/iotdb-system.properties.template | 27 ++ .../iotdb/commons/service/metric/enums/Metric.java | 5 + 17 files changed, 1170 insertions(+), 122 deletions(-) diff --git a/docs/zh/UserGuide/Subscription/Memory-Management.md b/docs/zh/UserGuide/Subscription/Memory-Management.md new file mode 100644 index 00000000000..eab583e060c --- /dev/null +++ b/docs/zh/UserGuide/Subscription/Memory-Management.md @@ -0,0 +1,91 @@ +<!-- + 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. +--> + +# 大字段与大 batch 订阅的内存配置 + +本页描述支持队列 fair-share 内存准入的版本。实际部署前，请确认所用版本包含这一行为；仅调整参数不能为旧版本增加队列隔离。 + +## 区分物化预算与响应缓存 + +| 参数 | 控制范围 | 调整建议 | +| --- | --- | --- | +| `chunk_timeseriesmeta_free_memory_proportion` 第九项 `Subscription` | 每个 DataNode 上，consensus 订阅队列物化并保留的 tablet 内存 | 按最大 entry、活跃队列数及缓冲深度计算预算；与其它查询内存份额一起调整 | +| `subscription_cache_memory_usage_percentage` | 序列化 poll 响应缓存的上限，占 pipe non-floating 内存的比例 | 从默认 `0.2` 开始；响应频繁淘汰和重复序列化、且 pipe 内存充足时，压测比较 `0.3`、`0.4` | +| `subscription_consensus_batch_max_size_in_bytes` | 物化后合并投递 batch 的目标大小 | 默认 8 MiB 不能限制单条 WAL entry 的大小；需要更小的 entry 时，应减小写入 batch 或字段大小 | +| `subscription_consensus_prefetch_executor_max_thread_num` | 同节点所有队列共用的预取线程池 | 大字段场景可比较 2 与 4 线程，观察 CPU、堆内存、ACK 延迟和普通队列吞吐 | + +`subscription_cache_memory_usage_percentage` 不控制 tablet 物化预算。其上限为 `pipe non-floating 内存 × 配置比例`；缓存实际获得的额度可能因 pipe 内存压力而更小。增加该比例不能解除物化额度阻塞，也不宜在未观察缓存淘汰的情况下统一调高。 + +第九项控制的物化预算为： + +```text +节点订阅物化预算 = query 内存预算 × Subscription 权重 / 九项权重总和 +``` + +例如 `1:100:200:50:200:200:200:50:250` 中，Subscription 比例为 `250 / 1251`，约 20%。修改任何一项都会影响其它项的实际比例。这些配置需要重启生效。 + +## 计算物化预算 + +计数单位是同一 DataNode 上活跃的订阅队列。每个 consumer group/topic/DataRegion 组合有独立队列；一个全量 topic、一个 consumer group 也可能对应多条队列，增加同组 consumer 数不会使每个 consumer 单独获得一份物化额度。 + +以 `E` 表示最大单条 WAL entry 转换得到的所有 tablet 的估算内存总和，`Q` 表示活跃队列数，`K` 表示希望每条队列同时保留的 entry 数量。初始预算可按以下公式估算，再用混合负载压测调整： + +```text +节点订阅物化预算 >= E × Q × K × 1.2 +``` + +`K=1` 是最小容量估算；需要在等待 ACK 时继续物化，可从 `K=2` 开始。20% 余量是容量估算起点，并非整个 JVM 的内存保证。写入 request、序列化响应缓存及转换中的临时对象还会占用其它内存；`E` 应使用物化后的内存估算，不能直接套用网络帧大小。 + +例如，最大物化 entry 为 60 MB、4 条活跃队列时，`K=1` 的估算为 288 MB，`K=2` 为 576 MB。如果节点已有约 987 MB 的物化预算，则每条队列约有 247 MB。这个容量本身可以容纳 30–60 MB 级 entry，无需仅因实时拒收就提高缓存比例；应先核实拒收原因、ACK 速度和物化估算大小。权重对应的内存量还取决于实际 query 内存预算，不能将固定权重当成固定 MB 数。 + +## 准入、错误及恢复 + +活跃队列均分节点物化预算。队列达到自己的额度后停止接收新的实时引用，已在 pending 队列中的 entry 会保留；未接收的实时 entry 仍可通过保留的 WAL 回放。已物化的数据可以继续投递，并在 ACK 后释放额度。 + +新队列激活后，配额会重新计算。旧队列已分配的内存通过投递和 ACK 逐步释放，因此重新分配额度可能出现短暂等待。队列停用时，在其保留内存释放之前不会把这份额度提前分给其它队列。 + +单条 entry 超过该队列整个额度时，明确拒绝物化，并在队列已可投递的数据处理完后返回带 `SUBSCRIPTION_OVERSIZED_ENTRY` 的 critical `ERROR` poll 响应。此时不会跳过 entry，也不会推进其 WAL/提交进度。对于历史积压，应增加物化预算、减少并发队列或调整 topic 范围后重新消费；减小未来的写入 batch 不会缩小已经写入 WAL 的 entry。重放仍受 WAL 保留策略约束。 + +拒收日志包含 `reasonCode`、`group`、`queueSize` 和 `queueRemaining`，文案说明实时准入拒绝与 WAL 补投。常见原因如下： + +| `reasonCode` | 含义 | +| --- | --- | +| `SUBSCRIPTION_MEMORY_QUOTA` | 队列已有保留数据，本次物化或实时准入超过自身可用额度 | +| `SUBSCRIPTION_MEMORY_LIMIT` | 物化时节点额度暂时不足，例如新队列激活后旧分配尚未释放 | +| `SUBSCRIPTION_OVERSIZED_ENTRY` | 单条物化 entry 超过队列整个额度，等待 ACK 也无法使它装入 | +| `SUBSCRIPTION_WRITER_BACKLOG` | 已物化的数据等待 writer 顺序条件满足 | +| `SUBSCRIPTION_QUEUE_CAPACITY` | pending 队列条数达到容量上限 | +| `CONSENSUS_REQUEST_MEMORY_LIMIT` | 实时 request 的 consensus 内存预留失败 | +| `SUBSCRIPTION_QUEUE_INACTIVE_OR_CLOSED` / `SUBSCRIPTION_SEEK_IN_PROGRESS` | 队列停用、关闭或正在 seek | + +## 监控及压测验收 + +以下指标按 `name`（consumer group/topic）和 `region` 区分队列，需启用指标采集： + +| 指标 | 含义 | +| --- | --- | +| `subscription_consensus_memory_used` | 队列保留的物化内存字节数 | +| `subscription_consensus_memory_quota` | 当前队列物化额度字节数 | +| `subscription_consensus_admission_rejections` | 队列生命周期内实时 offer 被拒绝的累计次数 | +| `subscription_consensus_memory_rejections` | 内存额度/节点预算导致 offer 或物化分配失败的累计次数 | +| `subscription_consensus_oversized_rejections` | 超大 entry 导致 offer 或物化分配失败的累计次数 | + +拒收累计值不是永久丢失计数。应结合消费进度、WAL lag、投递速率及 ACK 延迟判断积压能否追平。监控采集可对累计值计算增量；队列重建后累计值会重新开始。 + +大字段和普通数值写入混跑时，应确认大帧队列达到自身额度后，普通队列仍能持续投递。停止写入后应等待已知写入目标全部消费完成，并核对源端查询量、写入成功量及消费量；单纯以一段时间无 message 作为完成判据，可能在写入仍进行时提前退出。 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..4e582c79824 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 @@ -39,13 +39,13 @@ 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(); + private final AtomicLong rejectedEntries = new AtomicLong(); + private final AtomicLong lastRejectionLogTimeMs = new AtomicLong(); public SubscriptionQueueRegistry(final String consensusGroupId) { this.consensusGroupId = consensusGroupId; @@ -141,27 +141,33 @@ 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 long rejectedCount = rejectedEntries.incrementAndGet(); final long now = System.currentTimeMillis(); - final long lastLogTime = lastDropLogTimeMs.get(); - if (now - lastLogTime >= QUEUE_FULL_LOG_INTERVAL_MS - && lastDropLogTimeMs.compareAndSet(lastLogTime, now)) { + final long lastLogTime = lastRejectionLogTimeMs.get(); + if (now - lastLogTime >= REJECTION_LOG_INTERVAL_MS + && lastRejectionLogTimeMs.compareAndSet(lastLogTime, now)) { 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, + rejectedEntries.getAndSet(0), + 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 1c0362b627b..dfbd4e42b24 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 @@ -2645,4 +2645,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_EXCEEDING_QUEUE_QUOTA_ARG_BYTES_DATANODE_BUDGET_ARG_BYTES_REDUCE_THE_WRITE_BATCH_FIELD_SIZE_OR_INCREASE_THE_SUBSCRIPTION_PROPORTION_IN_CHUNK_TIMESERIESMETA_FREE_MEMORY_PROPORTION_WAL_PROGRESS_HAS_NOT_ADVANCED_C9120068 = + "[%s] Subscription entry requires %d bytes, exceeding queue quota %d bytes (DataNode budget %d bytes). Reduce the write batch/field size or increase the Subscription proportion in chunk_timeseriesmeta_free_memory_proportion. 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 49096659ade..744c887bd7b 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_EXCEEDING_QUEUE_QUOTA_ARG_BYTES_DATANODE_BUDGET_ARG_BYTES_REDUCE_THE_WRITE_BATCH_FIELD_SIZE_OR_INCREASE_THE_SUBSCRIPTION_PROPORTION_IN_CHUNK_TIMESERIESMETA_FREE_MEMORY_PROPORTION_WAL_PROGRESS_HAS_NOT_ADVANCED_C9120068 = + "[%s] 订阅 entry 需要 %d 字节，超过队列配额 %d 字节（DataNode 预算 %d 字节）。请减小写入 batch/字段大小，或提高 chunk_timeseriesmeta_free_memory_proportion 中的 Subscription 比例。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 b692ec94ed0..8efff2b6cf4 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; @@ -95,7 +97,7 @@ import java.util.concurrent.atomic.AtomicLong; 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 +153,24 @@ 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 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 +435,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 +488,7 @@ public class ConsensusPrefetchingQueue { retainedRequestBytes.addAndGet(request.getRetainedMemorySize()); } else { requestMemoryManager.free(request); + reject(SubscriptionQueueRejectionReason.QUEUE_CAPACITY); } } if (offered) { @@ -478,6 +517,11 @@ public class ConsensusPrefetchingQueue { private long getRetainedRequestBytes() { return retainedRequestBytes.get(); } + + @Override + public SubscriptionQueueRejectionReason getLastRejectionReason() { + return lastOfferRejectionReason.get(); + } } private static final class PendingSeekRequest { @@ -559,6 +603,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 +623,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 +649,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 +693,52 @@ 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 SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_QUOTA; + } + 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(); + 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 +858,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_EXCEEDING_QUEUE_QUOTA_ARG_BYTES_DATANODE_BUDGET_ARG_BYTES_REDUCE_THE_WRITE_BATCH_FIELD_SIZE_OR_INCREASE_THE_SUBSCRIPTION_PROPORTION_IN_CHUNK_TIMESERIESMETA_FREE_MEMORY_PROPORTION_WAL_PROGRESS_HAS_NOT_ADVANCED_C9120068, + SubscriptionQueueRejectionReason.SUBSCRIPTION_OVERSIZED_ENTRY.getCode(), + blockedEntryBytes, + subscriptionMemoryHandle.getMemoryQuotaInBytes(), + subscriptionMemoryManager.getTotalMemorySizeInBytes()), + true), + createNonCommittableContext(IoTDBDescriptor.getInstance().getConfig().getDataNodeId())); + } return event; } finally { releaseReadLock(); @@ -1428,7 +1529,7 @@ public class ConsensusPrefetchingQueue { recycleInFlightEvents(); if (!isActive) { - blockRealtimeAdmission(); + blockRealtimeAdmission(SubscriptionQueueRejectionReason.INACTIVE_OR_CLOSED); return computeIdleRoundResult(); } if (prefetchingQueue.size() >= MAX_PREFETCHING_QUEUE_SIZE) { @@ -1451,12 +1552,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(); @@ -1501,7 +1602,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(); @@ -1516,7 +1617,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) { @@ -2321,11 +2422,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; } @@ -2333,22 +2442,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 = SubscriptionQueueRejectionReason.SUBSCRIPTION_MEMORY_QUOTA; + return true; + } + 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; } @@ -2366,7 +2474,20 @@ 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 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() { @@ -3032,7 +3153,7 @@ public class ConsensusPrefetchingQueue { // ======================== Cleanup ======================== public void cleanUp() { - blockRealtimeAdmission(); + blockRealtimeAdmission(SubscriptionQueueRejectionReason.INACTIVE_OR_CLOSED); acquireWriteLock(); try { prefetchingQueue.forEach(event -> cleanUpEvent(event, true)); @@ -3048,7 +3169,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; @@ -3059,6 +3181,8 @@ public class ConsensusPrefetchingQueue { pendingSubscriptionWalResetSearchIndex = Long.MIN_VALUE; pendingSubscriptionWalResetGeneration = Long.MIN_VALUE; closeSubscriptionWALIterator(); + subscriptionMemoryHandle.setActive(false); + subscriptionMemoryHandle.close(); } finally { releaseWriteLock(); @@ -3317,7 +3441,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)); @@ -3327,6 +3451,7 @@ public class ConsensusPrefetchingQueue { discardRetainedEventMemory(); // 2. Discard stale pending entries from in-memory queue + pendingEntries.clear(); memoryBlockedEntryBytes = -1L; unblockRealtimeAdmission(); @@ -4038,12 +4163,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; @@ -4394,6 +4521,30 @@ public class ConsensusPrefetchingQueue { return subscriptionMemoryManager.getTotalMemorySizeInBytes(); } + public long getSubscriptionMemoryUsedInBytes() { + return subscriptionMemoryHandle.getUsedMemorySizeInBytes(); + } + + public long getSubscriptionMemoryQuotaInBytes() { + return subscriptionMemoryHandle.getMemoryQuotaInBytes(); + } + + public long getRealtimeAdmissionRejectionCount() { + return realtimeAdmissionRejectionCount.get(); + } + + 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(); @@ -4471,6 +4622,15 @@ public class ConsensusPrefetchingQueue { result.put( "subscriptionMemoryLimitInBytes", String.valueOf(subscriptionMemoryManager.getTotalMemorySizeInBytes())); + result.put("subscriptionMemoryUsedInBytes", String.valueOf(getSubscriptionMemoryUsedInBytes())); + result.put( + "subscriptionMemoryQuotaInBytes", String.valueOf(getSubscriptionMemoryQuotaInBytes())); + 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..36eb3299f5f 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 @@ -199,6 +199,36 @@ 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_ADMISSION_REJECTIONS.toString(), + MetricLevel.IMPORTANT, + queue, + ConsensusPrefetchingQueue::getRealtimeAdmissionRejectionCount, + key.getTags()); + 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 +339,24 @@ 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_ADMISSION_REJECTIONS.toString(), + key.getTags()); + 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) { 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..bf51ab9a32c 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,28 @@ 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 and receive an equal share of the configured budget. + * A queue cannot consume another queue's share, while an entry larger than a share is rejected with + * an explicit reason, even when the node budget is otherwise empty. Quotas change as queues become + * active or release their retained data; existing allocations are drained rather than revoked. + */ public class SubscriptionMemoryManager { private static final String MEMORY_BLOCK_NAME = "Subscription"; private final IMemoryBlock memoryBlock; + 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 +63,300 @@ 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(), + Math.min( + handle.getMemoryQuotaInBytesUnderLock() - handle.usedMemoryInBytes, + 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(), + getFreeMemorySizeInBytes()); + } + + final long total = getTotalMemorySizeInBytes(); + final long used = getUsedMemorySizeInBytes(); + final long quota = handle.getMemoryQuotaInBytesUnderLock(); + final long ownerUsed = handle.usedMemoryInBytes; + final boolean fitsInNode = total > 0L && used <= total && sizeInBytes <= total - used; + final boolean fitsInShare = ownerUsed <= quota && sizeInBytes <= quota - ownerUsed; + + if (fitsInNode && fitsInShare) { + return AllocationResult.accepted(quota, quota - handle.usedMemoryInBytes); + } + + final AllocationRejectionReason reason; + if (sizeInBytes > quota) { + reason = AllocationRejectionReason.OVERSIZED_ENTRY; + } else if (!fitsInShare) { + reason = AllocationRejectionReason.MEMORY_QUOTA; + } else { + reason = AllocationRejectionReason.MEMORY_LIMIT; + } + return AllocationResult.rejected(reason, quota, Math.min(quota - 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 getFreeMemorySizeInBytes() { + synchronized (SubscriptionMemoryManager.this) { + // Realtime request admission uses only this queue's quota. Node-wide allocation checks + // happen during materialization, including when pre-existing allocations exceed new shares. + return Math.max(0L, getMemoryQuotaInBytesUnderLock() - usedMemoryInBytes); + } + } + + 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..190b795f7e3 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,187 @@ 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 = + new SubscriptionMemoryManager(largeBytes * 2L); + 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()); + + 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 +261,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 +306,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()); + assertEquals("true", queueB.coreReportMessage().get("realtimeAdmissionBlocked")); 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 +382,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 +436,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 +472,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 0f321341401..68df431cf63 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; @@ -635,7 +638,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); @@ -645,7 +649,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(); @@ -657,11 +661,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( @@ -714,6 +724,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 @@ -722,7 +803,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); @@ -734,10 +816,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 { @@ -749,6 +841,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(); @@ -988,6 +1083,7 @@ public class ConsensusPrefetchingQueueTest { 1L, 1L, true); + queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(1024L * 1024L)); final IndexedConsensusRequest dataRequest = new IndexedConsensusRequest( @@ -1760,6 +1856,7 @@ public class ConsensusPrefetchingQueueTest { 1L, 1L, true); + queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(1024L * 1024L)); final IndexedConsensusRequest dataRequest = new IndexedConsensusRequest( @@ -1828,6 +1925,7 @@ public class ConsensusPrefetchingQueueTest { 1L, 1L, true); + queue.setSubscriptionMemoryManager(new SubscriptionMemoryManager(1024L * 1024L)); final IndexedConsensusRequest dataRequest = new IndexedConsensusRequest( @@ -2636,6 +2734,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/resource/SubscriptionMemoryManagerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/resource/SubscriptionMemoryManagerTest.java index 1635a6e0ecf..087c5811776 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,79 @@ public class SubscriptionMemoryManagerTest { assertFalse(memoryManager.tryAllocate(1L)); assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); } + + @Test + public void testQueueCannotBorrowAnotherQueuesShare() { + 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(6L)); + assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + + assertFalse(queueA.tryAllocate(6L).isAccepted()); + assertTrue(queueA.tryAllocate(4L).isAccepted()); + assertEquals( + SubscriptionMemoryManager.AllocationRejectionReason.MEMORY_QUOTA, + queueA.inspectRejection(2L)); + assertTrue(queueB.tryAllocate(5L).isAccepted()); + assertTrue(queueA.tryAllocate(1L).isAccepted()); + assertEquals(10L, memoryManager.getUsedMemorySizeInBytes()); + + queueA.close(); + assertEquals(10L, queueB.getMemoryQuotaInBytes()); + assertTrue(queueB.tryAllocate(5L).isAccepted()); + assertEquals(10L, memoryManager.getUsedMemorySizeInBytes()); + + queueB.close(); + assertEquals(0L, memoryManager.getUsedMemorySizeInBytes()); + } + + @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..30d55945cb7 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,12 @@ 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. Active queues divide this budget equally. Size it +# 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 a queue's quota reports +# SUBSCRIPTION_OVERSIZED_ENTRY and does not advance WAL progress. Existing allocations drain via ACK +# when new queues reduce their quotas. # effectiveMode: restart chunk_timeseriesmeta_free_memory_proportion=1:100:200:50:200:200:200:50:250 @@ -2153,6 +2159,27 @@ 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. +# This does not control consensus subscription tablet materialization. Start with 0.2; for large +# responses, evaluate 0.3 or 0.4 only if cache eviction/reserialization is high and pipe memory has +# headroom. 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..31b6ff2820a 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,11 @@ 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_ADMISSION_REJECTIONS("subscription_consensus_admission_rejections"), + 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"),
