From 24870dc9d1a96bac7b06d70c44225867eb03c61e Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 14 Oct 2026 19:01:58 +0800 Subject: [PATCH 1/2] [Subscription] Support per-topic uncommitted event recycle timeouts --- .../rpc/subscription/config/TopicConfig.java | 8 ++ .../subscription/config/TopicConstant.java | 4 + .../subscription/config/TopicConfigTest.java | 15 ++ .../confignode/i18n/ConfigNodeMessages.java | 3 + .../confignode/i18n/ConfigNodeMessages.java | 3 + .../subscription/SubscriptionInfo.java | 30 ++++ .../SubscriptionInfoTopicValidationTest.java | 83 +++++++++++ .../agent/SubscriptionBrokerAgent.java | 10 ++ .../agent/SubscriptionTopicAgent.java | 47 +++++++ .../broker/ConsensusSubscriptionBroker.java | 8 ++ .../broker/SubscriptionPrefetchingQueue.java | 13 +- .../consensus/ConsensusPrefetchingQueue.java | 20 ++- .../subscription/event/SubscriptionEvent.java | 11 +- .../agent/SubscriptionTopicAgentTest.java | 109 +++++++++++++++ ...onPrefetchingQueueRecycleIntervalTest.java | 131 ++++++++++++++++++ .../ConsensusPrefetchingQueueTest.java | 108 +++++++++++++++ .../conf/iotdb-system.properties.template | 10 ++ 17 files changed, 601 insertions(+), 12 deletions(-) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueueRecycleIntervalTest.java diff --git a/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java b/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java index 51b87f910e2dd..ae371ccfd32de 100644 --- a/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java +++ b/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java @@ -190,6 +190,14 @@ public static boolean isValidOrderMode(final String orderMode) { return ORDER_MODE_VALUE_SET.contains(normalizeOrderMode(orderMode)); } + /** Returns the topic's recycle timeout in milliseconds, or null to use the server config. */ + public Long getRecycleUncommittedEventIntervalMs() { + return containsKeyIgnoreCase(TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY) + ? Long.parseLong( + getStringIgnoreCase(TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, "").trim()) + : null; + } + public static String normalizeOrderMode(final String orderMode) { return orderMode == null ? TopicConstant.ORDER_MODE_DEFAULT_VALUE diff --git a/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java b/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java index 960e3bab524b1..8a8af71d73889 100644 --- a/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java +++ b/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java @@ -88,6 +88,10 @@ public class TopicConstant { public static final String STRICT_KEY = "strict"; public static final String STRICT_DEFAULT_VALUE = "true"; + // Overrides subscription_recycle_uncommitted_event_interval_ms for this topic. + public static final String RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY = + "recycle-uncommitted-event-interval-ms"; + public static final String OWNER_ID_KEY = "owner-id"; public static final String OWNER_EPOCH_KEY = "owner-epoch"; // Highest owner epoch ever granted for this topic. Retained across clearOwner so that owner diff --git a/iotdb-client/subscription/src/test/java/org/apache/iotdb/rpc/subscription/config/TopicConfigTest.java b/iotdb-client/subscription/src/test/java/org/apache/iotdb/rpc/subscription/config/TopicConfigTest.java index 3b32d407eba94..a1ae258c8cae7 100644 --- a/iotdb-client/subscription/src/test/java/org/apache/iotdb/rpc/subscription/config/TopicConfigTest.java +++ b/iotdb-client/subscription/src/test/java/org/apache/iotdb/rpc/subscription/config/TopicConfigTest.java @@ -49,6 +49,21 @@ public void testCanonicalModeValues() { Assert.assertTrue(topicConfigWithMode(" INCREMENTAL ").isIncrementalMode()); } + @Test + public void testRecycleUncommittedEventInterval() { + Assert.assertNull(new TopicConfig().getRecycleUncommittedEventIntervalMs()); + + final Map attributes = new HashMap<>(); + attributes.put(TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, " 120000 "); + Assert.assertEquals( + Long.valueOf(120000L), new TopicConfig(attributes).getRecycleUncommittedEventIntervalMs()); + + attributes.clear(); + attributes.put("RECYCLE-UNCOMMITTED-EVENT-INTERVAL-MS", "300000"); + Assert.assertEquals( + Long.valueOf(300000L), new TopicConfig(attributes).getRecycleUncommittedEventIntervalMs()); + } + @SuppressWarnings("deprecation") @Test public void testLegacyModeValues() { diff --git a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java index fb73cc9337632..bffe196b708ba 100644 --- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java +++ b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java @@ -20,6 +20,9 @@ package org.apache.iotdb.confignode.i18n; public final class ConfigNodeMessages { + public static final String + EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ILLEGAL_ARG_ARG_EXPECTED_A_POSITIVE_LONG_VALUE_IN_MILLISECONDS_8A5C0D8C = + "Failed to create or alter topic, illegal %s=%s, expected a positive long value in milliseconds"; public static final String LOG_CONFIGREGION_UPDATE_CONFIGNODE_NODEID_ARG_CONSENSUSENDPOINT_ARG_IS_A_CONFIGREGION_FOLLOWER_NEW_LEADER_NODEID_ARG_STOPPING_LEADER_SERVICES_A1962CD2 = "[ConfigRegion Update] ConfigNode [nodeId: {}, consensusEndpoint: {}] is a ConfigRegion follower; new leader nodeId: {}. Stopping leader services."; diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java index 948f4a6476512..b1910b2c3b91e 100644 --- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java +++ b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java @@ -20,6 +20,9 @@ package org.apache.iotdb.confignode.i18n; public final class ConfigNodeMessages { + public static final String + EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ILLEGAL_ARG_ARG_EXPECTED_A_POSITIVE_LONG_VALUE_IN_MILLISECONDS_8A5C0D8C = + "创建或修改 topic 失败，%s=%s 无效，需指定以毫秒为单位的正 long 值"; public static final String LOG_CONFIGREGION_UPDATE_CONFIGNODE_NODEID_ARG_CONSENSUSENDPOINT_ARG_IS_A_CONFIGREGION_FOLLOWER_NEW_LEADER_NODEID_ARG_STOPPING_LEADER_SERVICES_A1962CD2 = "[ConfigRegion Update] ConfigNode [nodeId: {}, consensusEndpoint: {}] 为 ConfigRegion follower；新 leader 的 nodeId: {}。开始停止 leader 服务。"; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java index 9fab5d1f536b9..2796a0a1bf832 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java @@ -109,6 +109,7 @@ public class SubscriptionInfo implements SnapshotProcessor { TopicConstant.RETENTION_BYTES_KEY, TopicConstant.RETENTION_MS_KEY, TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY, + TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, TopicConstant.START_TIME_KEY, TopicConstant.END_TIME_KEY, TopicConstant.MODE_KEY, @@ -131,6 +132,7 @@ public class SubscriptionInfo implements SnapshotProcessor { TopicConstant.RETENTION_BYTES_KEY, TopicConstant.RETENTION_MS_KEY, TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY, + TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, TopicConstant.MODE_KEY, TopicConstant.ORDER_MODE_KEY, TopicConstant.FORMAT_KEY, @@ -399,6 +401,7 @@ private void validateTopicConfig(final TopicConfig topicConfig) throws Subscript validateColumnFilter(topicConfig); validateIncrementalTopicRetentionConfig(topicConfig); + validateRecycleUncommittedEventInterval(topicConfig); final Long ownerLeaseDurationMs = topicConfig.getLong(TopicConstant.OWNER_LEASE_DURATION_MS_KEY); @@ -586,6 +589,33 @@ private boolean isIncrementalTopicConfig(final TopicConfig topicConfig) { return topicConfig.isIncrementalMode(); } + private void validateRecycleUncommittedEventInterval(final TopicConfig topicConfig) + throws SubscriptionException { + try { + final Long intervalMs = topicConfig.getRecycleUncommittedEventIntervalMs(); + if (Objects.isNull(intervalMs) || intervalMs > 0L) { + return; + } + } catch (final NumberFormatException ignored) { + // Report malformed values together with non-positive timeouts below. + } + final String exceptionMessage = + String.format( + ConfigNodeMessages + .EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ILLEGAL_ARG_ARG_EXPECTED_A_POSITIVE_LONG_VALUE_IN_MILLISECONDS_8A5C0D8C, + TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, + topicConfig.getAttribute().entrySet().stream() + .filter( + entry -> + TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY.equalsIgnoreCase( + entry.getKey())) + .map(entry -> String.valueOf(entry.getValue())) + .findFirst() + .orElse("")); + LOGGER.warn(exceptionMessage); + throw new SubscriptionException(exceptionMessage); + } + private void validateIncrementalTopicRetentionConfig(final TopicConfig topicConfig) throws SubscriptionException { if (!hasTopicAttributeIgnoreCase(topicConfig, TopicConstant.RETENTION_BYTES_KEY) diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTopicValidationTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTopicValidationTest.java index da9c356190c64..5b15b7af7f34d 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTopicValidationTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTopicValidationTest.java @@ -305,6 +305,89 @@ public void testRejectIllegalRetentionFormat() { assertCreateRejected(subscriptionInfo, attributes, "expected a long value"); } + @Test + public void testAcceptRecycleUncommittedEventInterval() throws Exception { + final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); + + final Map initialAttributes = newInitialTableTopicAttributes(); + initialAttributes.put( + TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, " 9223372036854775807 "); + Assert.assertTrue( + subscriptionInfo.validateBeforeCreatingTopic( + new TCreateTopicReq("initial_topic").setTopicAttributes(initialAttributes))); + + final Map incrementalAttributes = newIncrementalTableTopicAttributes(); + incrementalAttributes.put("Recycle-Uncommitted-Event-Interval-Ms", "300000"); + Assert.assertTrue( + subscriptionInfo.validateBeforeCreatingTopic( + new TCreateTopicReq("incremental_topic").setTopicAttributes(incrementalAttributes))); + + for (final String format : + new String[] { + TopicConstant.FORMAT_TS_FILE_VALUE, TopicConstant.FORMAT_RECORD_HANDLER_VALUE + }) { + final Map snapshotAttributes = new HashMap<>(); + snapshotAttributes.put(TopicConstant.MODE_KEY, TopicConstant.MODE_SNAPSHOT_VALUE); + snapshotAttributes.put(TopicConstant.FORMAT_KEY, format); + snapshotAttributes.put(TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, "1200000"); + Assert.assertTrue( + subscriptionInfo.validateBeforeCreatingTopic( + new TCreateTopicReq("snapshot_topic").setTopicAttributes(snapshotAttributes))); + } + } + + @Test + public void testRejectInvalidRecycleUncommittedEventInterval() { + final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); + for (final String value : + new String[] {"0", "-1", "not-a-number", "", " ", "1.5", "9223372036854775808", null}) { + final Map attributes = newInitialTableTopicAttributes(); + attributes.put(TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, value); + assertCreateRejected( + subscriptionInfo, attributes, "expected a positive long value in milliseconds"); + } + } + + @Test + public void testAcceptAlteringRecycleUncommittedEventInterval() throws Exception { + final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); + for (final String mode : + new String[] {TopicConstant.MODE_INITIAL_VALUE, TopicConstant.MODE_INCREMENTAL_VALUE}) { + final Map attributes = newInitialTableTopicAttributes(); + attributes.put(TopicConstant.MODE_KEY, mode); + attributes.put(TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, "1200000"); + subscriptionInfo.createTopic(new CreateTopicPlan(new TopicMeta(mode, 1L, attributes))); + + final Map updatedAttributes = new HashMap<>(attributes); + updatedAttributes.put(TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, "300000"); + subscriptionInfo.validateBeforeAlteringTopic(new TopicMeta(mode, 1L, updatedAttributes)); + } + } + + @Test + public void testRejectAlteringInvalidRecycleUncommittedEventInterval() { + final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); + final Map attributes = newInitialTableTopicAttributes(); + subscriptionInfo.createTopic(new CreateTopicPlan(new TopicMeta("table_topic", 1L, attributes))); + attributes.put(TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, "0"); + try { + subscriptionInfo.validateBeforeAlteringTopic(new TopicMeta("table_topic", 1L, attributes)); + Assert.fail("Expected an invalid recycle interval to be rejected"); + } catch (final SubscriptionException e) { + Assert.assertTrue(e.getMessage().contains("expected a positive long value in milliseconds")); + } + } + + @Test + public void testRejectDuplicateRecycleUncommittedEventIntervals() { + final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); + final Map attributes = newInitialTableTopicAttributes(); + attributes.put(TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, "1200000"); + attributes.put("RECYCLE-UNCOMMITTED-EVENT-INTERVAL-MS", "300000"); + assertCreateRejected( + subscriptionInfo, attributes, "duplicate recycle-uncommitted-event-interval-ms attributes"); + } + @Test public void testRejectAlteringRetentionConfig() throws Exception { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java index 1c429cd97dca3..aafcf903dc43e 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java @@ -719,6 +719,16 @@ public void refreshConsensusQueueOrderMode( } } + public void wakeUpConsensusPrefetchingQueues(final String topicName, final boolean isTableModel) { + for (final List brokers : consumerGroupIdToBrokers.values()) { + final ConsensusSubscriptionBroker broker = + getBroker(brokers, ConsensusSubscriptionBroker.class); + if (Objects.nonNull(broker)) { + broker.wakeUpPrefetchingQueues(topicName, isTableModel); + } + } + } + public void refreshColumnFilter(final String topicName, final TopicConfig topicConfig) { final ColumnFilterMatcher matcher; try { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java index f4cb98d823ed7..f68e60f353f8a 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgent.java @@ -26,6 +26,7 @@ import org.apache.iotdb.commons.pipe.datastructure.pattern.IoTDBTreePattern; import org.apache.iotdb.commons.pipe.datastructure.pattern.PrefixTreePattern; import org.apache.iotdb.commons.pipe.datastructure.pattern.TreePattern; +import org.apache.iotdb.commons.subscription.config.SubscriptionConfig; import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta; import org.apache.iotdb.commons.subscription.meta.topic.TopicMetaKeeper; import org.apache.iotdb.db.auth.AuthorityChecker; @@ -40,14 +41,17 @@ import org.apache.iotdb.rpc.subscription.config.TopicConfig; import org.apache.iotdb.rpc.subscription.config.TopicConstant; +import org.apache.tsfile.utils.Pair; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.util.HashSet; import java.util.List; import java.util.Locale; import java.util.Map; import java.util.Objects; import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; import java.util.stream.Collectors; public class SubscriptionTopicAgent { @@ -56,6 +60,12 @@ public class SubscriptionTopicAgent { private final TopicMetaKeeper topicMetaKeeper; + // Queue operations read this derived runtime config without acquiring the topic-meta lock. + // Topic-meta updates can wake queues while holding that lock, so taking it from a queue operation + // would reverse the lock order. The data model is part of the key to isolate same-name topics. + private final Map, Long> topicRecycleIntervalsMs = + new ConcurrentHashMap<>(); + public SubscriptionTopicAgent() { this.topicMetaKeeper = new TopicMetaKeeper(); } @@ -109,8 +119,18 @@ private void handleSingleTopicMetaChangesInternal(final TopicMeta metaFromCoordi final boolean isTableModel = metaFromCoordinator.visibleUnderTableModel(); final TopicMeta oldMeta = topicMetaKeeper.getTopicMeta(topicName, isTableModel); TopicMeta.validateOwnerProgression(oldMeta, metaFromCoordinator); + final Long recycleIntervalMs = + metaFromCoordinator.getConfig().getRecycleUncommittedEventIntervalMs(); topicMetaKeeper.removeTopicMeta(topicName, isTableModel); topicMetaKeeper.addTopicMeta(topicName, metaFromCoordinator); + final Pair topicKey = new Pair<>(topicName, isTableModel); + final Long oldRecycleIntervalMs = + Objects.isNull(recycleIntervalMs) + ? topicRecycleIntervalsMs.remove(topicKey) + : topicRecycleIntervalsMs.put(topicKey, recycleIntervalMs); + if (!Objects.equals(oldRecycleIntervalMs, recycleIntervalMs)) { + SubscriptionAgent.broker().wakeUpConsensusPrefetchingQueues(topicName, isTableModel); + } if (shouldRefreshColumnFilter(oldMeta, metaFromCoordinator)) { SubscriptionAgent.broker().refreshColumnFilter(topicName, metaFromCoordinator.getConfig()); } else if (!metaFromCoordinator.getConfig().isTableTopic() @@ -173,9 +193,14 @@ public TPushTopicMetaRespExceptionMessage handleTopicMetaChanges( final List topicMetasFromCoordinator) { acquireWriteLock(); try { + final Set> currentTopics = new HashSet<>(); for (final TopicMeta topicMetaFromCoordinator : topicMetasFromCoordinator) { try { handleSingleTopicMetaChangesInternal(topicMetaFromCoordinator); + currentTopics.add( + new Pair<>( + topicMetaFromCoordinator.getTopicName(), + topicMetaFromCoordinator.visibleUnderTableModel())); } catch (final Exception e) { final String topicName = topicMetaFromCoordinator.getTopicName(); LOGGER.warn( @@ -191,6 +216,15 @@ public TPushTopicMetaRespExceptionMessage handleTopicMetaChanges( topicName, exceptionMessage, System.currentTimeMillis()); } } + // Remove absent topics only after applying the full snapshot. Clearing first would let queue + // operations temporarily use the global timeout and could recycle an event prematurely. + for (final Pair topicKey : topicRecycleIntervalsMs.keySet()) { + if (!currentTopics.contains(topicKey)) { + topicRecycleIntervalsMs.remove(topicKey); + SubscriptionAgent.broker() + .wakeUpConsensusPrefetchingQueues(topicKey.left, topicKey.right); + } + } return null; } finally { releaseWriteLock(); @@ -238,6 +272,19 @@ private void handleDropTopicInternal(final String topicName, final Boolean isTab if (Objects.nonNull(topicMeta) && topicMeta.visibleUnderTableModel()) { SubscriptionAgent.broker().dropColumnFilter(topicName); } + if (Objects.nonNull(topicMeta) || Objects.nonNull(isTableModel)) { + topicRecycleIntervalsMs.remove( + new Pair<>( + topicName, + Objects.nonNull(isTableModel) ? isTableModel : topicMeta.visibleUnderTableModel())); + } + } + + public long getTopicRecycleUncommittedEventIntervalMs( + final String topicName, final boolean isTableModel) { + return topicRecycleIntervalsMs.getOrDefault( + new Pair<>(topicName, isTableModel), + (long) SubscriptionConfig.getInstance().getSubscriptionRecycleUncommittedEventIntervalMs()); } public boolean isTopicExisted(final String topicName) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ConsensusSubscriptionBroker.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ConsensusSubscriptionBroker.java index cd7abf043cf21..60b268504a2f9 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ConsensusSubscriptionBroker.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/ConsensusSubscriptionBroker.java @@ -751,6 +751,14 @@ public void unbindConsensusPrefetchingQueue( closeAndRemoveConsensusPrefetchingQueues(topicName, true, retainProgressAfterUnsubscribe); } + public void wakeUpPrefetchingQueues(final String topicName, final boolean isTableModel) { + final List queues = + topicNameToConsensusPrefetchingQueues.get(topicName); + if (Objects.nonNull(queues)) { + queues.forEach(queue -> queue.wakeUpForTopicConfigChange(isTableModel)); + } + } + @Override public void unbind(final String topicName) { unbindConsensusPrefetchingQueue(topicName, false); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java index 716e7d90f152f..2756f4fa4a974 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueue.java @@ -70,6 +70,7 @@ public abstract class SubscriptionPrefetchingQueue { private final String brokerId; // consumer group id private final String topicName; + private final boolean tableModel; /** Contains events coming from the upstream pipeline, corresponding to the pipe sink stage. */ private final SubscriptionBlockingPendingQueue inputPendingQueue; @@ -129,6 +130,7 @@ public SubscriptionPrefetchingQueue( final long maxBatchSizeInBytes) { this.brokerId = brokerId; this.topicName = topicName; + this.tableModel = SubscriptionAgent.consumer().isTableModel(brokerId); this.inputPendingQueue = inputPendingQueue; this.commitIdGenerator = commitIdGenerator; this.initialCommitId = commitIdGenerator.get(); @@ -157,6 +159,11 @@ public String getConsumerGroupId() { return brokerId; } + private long getRecycleUncommittedEventIntervalMs() { + return SubscriptionAgent.topic() + .getTopicRecycleUncommittedEventIntervalMs(topicName, tableModel); + } + protected void cleanUpInternal() { // clean up events in batches batches.cleanUp(); @@ -329,7 +336,7 @@ private SubscriptionEvent pollInternalV2(final String consumerId) { continue; } - if (!event.pollable()) { + if (!event.pollable(getRecycleUncommittedEventIntervalMs())) { LOGGER.warn( DataNodePipeMessages .PIPE_LOG_SUBSCRIPTION_SUBSCRIPTIONPREFETCHINGQUEUE_POLL_NON_POLLABLE_644D5D6B, @@ -373,7 +380,7 @@ private synchronized SubscriptionEvent pollPrefetchedEvent(final String consumer continue; } - if (!event.pollable()) { + if (!event.pollable(getRecycleUncommittedEventIntervalMs())) { LOGGER.warn( DataNodePipeMessages .PIPE_LOG_SUBSCRIPTION_SUBSCRIPTIONPREFETCHINGQUEUE_POLL_NON_POLLABLE_644D5D6B, @@ -1232,7 +1239,7 @@ private static RemappingFunction COMBINER( prefetchEvent(ev); // no need to log warn for eagerly pollable event return null; // remove this entry - } else if (ev.pollable()) { + } else if (ev.pollable(getRecycleUncommittedEventIntervalMs())) { ev.nack(); // now pollable if (ev.isPoisoned()) { LOGGER.error( 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 c19420c130cc7..90e6df8ec1883 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 @@ -750,6 +750,17 @@ private void requestPrefetch() { } } + public void wakeUpForTopicConfigChange(final boolean isTableModel) { + if (tableModel == isTableModel) { + requestPrefetch(); + } + } + + private long getRecycleUncommittedEventIntervalMs() { + return SubscriptionAgent.topic() + .getTopicRecycleUncommittedEventIntervalMs(topicName, tableModel); + } + private void requestPrefetchForRealtimeEntry() { if (prefetchingQueue.size() < MAX_PREFETCHING_QUEUE_SIZE) { requestPrefetch(); @@ -1491,7 +1502,7 @@ private SubscriptionEvent pollInternal(final String consumerId) { continue; } - if (!event.pollable()) { + if (!event.pollable(getRecycleUncommittedEventIntervalMs())) { LOGGER.warn( DataNodePipeMessages .PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_POLL_NON_POLLABLE_EVENT_BROKEN_E9551325, @@ -1920,10 +1931,7 @@ private PrefetchRoundResult computeIdleRoundResult( } if (!inFlightEvents.isEmpty()) { - delayMs = - Math.min( - delayMs, - SubscriptionConfig.getInstance().getSubscriptionRecycleUncommittedEventIntervalMs()); + delayMs = Math.min(delayMs, getRecycleUncommittedEventIntervalMs()); } return delayMs == Long.MAX_VALUE @@ -3399,7 +3407,7 @@ private void recycleInFlightEvents() { cleanUpEvent(ev, false); return null; } - if (ev.pollable()) { + if (ev.pollable(getRecycleUncommittedEventIntervalMs())) { ev.nack(); if (ev.isPoisoned()) { LOGGER.error( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/SubscriptionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/SubscriptionEvent.java index 2e58a68411da7..d055c2855a113 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/SubscriptionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/SubscriptionEvent.java @@ -235,13 +235,18 @@ public void recordLastPolledTimestamp() { * are not pollable. */ public boolean pollable() { + return pollable( + SubscriptionConfig.getInstance().getSubscriptionRecycleUncommittedEventIntervalMs()); + } + + public boolean pollable(final long recycleUncommittedEventIntervalMs) { if (isCommitted()) { return false; } if (lastPolledTimestamp.get() == INVALID_TIMESTAMP) { return true; } - return canRecycle(); + return canRecycle(recycleUncommittedEventIntervalMs); } /** @@ -255,11 +260,11 @@ public boolean eagerlyPollable() { return lastPolledTimestamp.get() == INVALID_TIMESTAMP; } - private boolean canRecycle() { + private boolean canRecycle(final long recycleUncommittedEventIntervalMs) { // Recycle events that may not be able to be committed, i.e., those that have been polled but // not committed within a certain period of time. return System.currentTimeMillis() - lastPolledTimestamp.get() - > SubscriptionConfig.getInstance().getSubscriptionRecycleUncommittedEventIntervalMs(); + > recycleUncommittedEventIntervalMs; } public void nack() { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgentTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgentTest.java index c31535b5812fc..1a658fdf1da83 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgentTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgentTest.java @@ -19,6 +19,8 @@ package org.apache.iotdb.db.subscription.agent; +import org.apache.iotdb.commons.pipe.config.constant.SystemConstant; +import org.apache.iotdb.commons.subscription.config.SubscriptionConfig; import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta; import org.apache.iotdb.rpc.subscription.config.TopicConstant; @@ -26,10 +28,117 @@ import org.junit.Test; import java.util.HashMap; +import java.util.List; import java.util.Map; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.spy; + public class SubscriptionTopicAgentTest { + @Test + public void testRecycleIntervalUpdateAndFallback() { + final SubscriptionTopicAgent agent = new SubscriptionTopicAgent(); + final long defaultInterval = + SubscriptionConfig.getInstance().getSubscriptionRecycleUncommittedEventIntervalMs(); + Assert.assertEquals( + defaultInterval, agent.getTopicRecycleUncommittedEventIntervalMs("recycle_topic", false)); + + Assert.assertNull(agent.handleSingleTopicMetaChanges(createRecycleTopicMeta(false, "1200000"))); + Assert.assertEquals( + 1200000L, agent.getTopicRecycleUncommittedEventIntervalMs("recycle_topic", false)); + + Assert.assertNull(agent.handleSingleTopicMetaChanges(createRecycleTopicMeta(false, "300000"))); + Assert.assertEquals( + 300000L, agent.getTopicRecycleUncommittedEventIntervalMs("recycle_topic", false)); + + Assert.assertNull(agent.handleSingleTopicMetaChanges(createRecycleTopicMeta(false, null))); + Assert.assertEquals( + defaultInterval, agent.getTopicRecycleUncommittedEventIntervalMs("recycle_topic", false)); + } + + @Test + public void testRecycleIntervalIsolationAndDrop() { + final SubscriptionTopicAgent agent = new SubscriptionTopicAgent(); + final long defaultInterval = + SubscriptionConfig.getInstance().getSubscriptionRecycleUncommittedEventIntervalMs(); + Assert.assertNull(agent.handleSingleTopicMetaChanges(createRecycleTopicMeta(false, "1200000"))); + Assert.assertNull(agent.handleSingleTopicMetaChanges(createRecycleTopicMeta(true, "300000"))); + + Assert.assertEquals( + 1200000L, agent.getTopicRecycleUncommittedEventIntervalMs("recycle_topic", false)); + Assert.assertEquals( + 300000L, agent.getTopicRecycleUncommittedEventIntervalMs("recycle_topic", true)); + + Assert.assertNull(agent.handleDropTopic("recycle_topic")); + Assert.assertEquals( + defaultInterval, agent.getTopicRecycleUncommittedEventIntervalMs("recycle_topic", false)); + Assert.assertEquals( + 300000L, agent.getTopicRecycleUncommittedEventIntervalMs("recycle_topic", true)); + Assert.assertNull(agent.handleDropTopic("recycle_topic", true)); + Assert.assertEquals( + defaultInterval, agent.getTopicRecycleUncommittedEventIntervalMs("recycle_topic", true)); + } + + @Test + public void testRecycleIntervalSnapshotPreservesCurrentOverride() { + final SubscriptionTopicAgent agent = new SubscriptionTopicAgent(); + final TopicMeta topicMeta = spy(createRecycleTopicMeta(false, "1200000")); + Assert.assertNull(agent.handleSingleTopicMetaChanges(topicMeta)); + doAnswer( + invocation -> { + Assert.assertEquals( + 1200000L, + agent.getTopicRecycleUncommittedEventIntervalMs("recycle_topic", false)); + return invocation.callRealMethod(); + }) + .when(topicMeta) + .getConfig(); + + Assert.assertNull(agent.handleTopicMetaChanges(List.of(topicMeta))); + } + + @Test + public void testRecycleIntervalSnapshotRemovesAbsentOverride() { + final SubscriptionTopicAgent agent = new SubscriptionTopicAgent(); + final long defaultInterval = + SubscriptionConfig.getInstance().getSubscriptionRecycleUncommittedEventIntervalMs(); + Assert.assertNull( + agent.handleTopicMetaChanges( + List.of( + createRecycleTopicMeta(false, "1200000"), createRecycleTopicMeta(true, "300000")))); + Assert.assertNull(agent.handleTopicMetaChanges(List.of(createRecycleTopicMeta(true, null)))); + + Assert.assertEquals( + defaultInterval, agent.getTopicRecycleUncommittedEventIntervalMs("recycle_topic", false)); + Assert.assertEquals( + defaultInterval, agent.getTopicRecycleUncommittedEventIntervalMs("recycle_topic", true)); + } + + @Test + public void testInvalidSnapshotPreservesExistingRecycleInterval() { + final SubscriptionTopicAgent agent = new SubscriptionTopicAgent(); + Assert.assertNull(agent.handleSingleTopicMetaChanges(createRecycleTopicMeta(false, "1200000"))); + Assert.assertNotNull( + agent.handleTopicMetaChanges(List.of(createRecycleTopicMeta(false, "invalid")))); + Assert.assertEquals( + 1200000L, agent.getTopicRecycleUncommittedEventIntervalMs("recycle_topic", false)); + } + + private static TopicMeta createRecycleTopicMeta( + final boolean tableModel, final String intervalMs) { + final Map attributes = new HashMap<>(); + attributes.put( + SystemConstant.SQL_DIALECT_KEY, + tableModel + ? SystemConstant.SQL_DIALECT_TABLE_VALUE + : SystemConstant.SQL_DIALECT_TREE_VALUE); + if (intervalMs != null) { + attributes.put(TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, intervalMs); + } + return new TopicMeta("recycle_topic", 1L, attributes); + } + @Test public void testColumnFilterRefreshSkippedForOwnerOnlyUpdate() { final TopicMeta oldMeta = createTableTopicMeta("column_name = \"temperature\"", "db", "t1"); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueueRecycleIntervalTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueueRecycleIntervalTest.java new file mode 100644 index 0000000000000..abebe01d51709 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueueRecycleIntervalTest.java @@ -0,0 +1,131 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.subscription.broker; + +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta; +import org.apache.iotdb.db.subscription.agent.SubscriptionAgent; +import org.apache.iotdb.db.subscription.event.SubscriptionEvent; +import org.apache.iotdb.rpc.subscription.config.TopicConstant; +import org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionCommitContext; +import org.apache.iotdb.rpc.subscription.payload.poll.SubscriptionPollResponseType; +import org.apache.iotdb.rpc.subscription.payload.poll.TerminationPayload; +import org.apache.iotdb.session.subscription.util.PollTimer; + +import org.junit.Test; + +import java.lang.reflect.Field; +import java.util.Collections; +import java.util.concurrent.atomic.AtomicLong; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class SubscriptionPrefetchingQueueRecycleIntervalTest { + + @Test + public void testRecycleIntervalForV1Queue() throws Exception { + verifyRecycleInterval(false); + } + + @Test + public void testRecycleIntervalForV2Queue() throws Exception { + verifyRecycleInterval(true); + } + + private static void verifyRecycleInterval(final boolean useV2) throws Exception { + final String topicName = "ordinary_recycle_topic_" + useV2; + final int originalInterval = + CommonDescriptor.getInstance() + .getConfig() + .getSubscriptionRecycleUncommittedEventIntervalMs(); + final SubscriptionBlockingPendingQueue inputQueue = + mock(SubscriptionBlockingPendingQueue.class); + when(inputQueue.isEmpty()).thenReturn(true); + final SubscriptionPrefetchingQueue queue = + new SubscriptionPrefetchingTabletQueue( + "recycle_group", topicName, inputQueue, new AtomicLong()); + try { + CommonDescriptor.getInstance() + .getConfig() + .setSubscriptionRecycleUncommittedEventIntervalMs(600000); + updateInterval(topicName, "1200000"); + final SubscriptionEvent event = + new SubscriptionEvent( + SubscriptionPollResponseType.TERMINATION.getType(), + new TerminationPayload(), + new SubscriptionCommitContext(1, 1, topicName, "recycle_group", 1)); + queue.prefetchEvent(event); + assertSame(event, poll(queue, useV2)); + + final Field timestampField = SubscriptionEvent.class.getDeclaredField("lastPolledTimestamp"); + timestampField.setAccessible(true); + ((AtomicLong) timestampField.get(event)).set(System.currentTimeMillis() - 900000L); + assertTrue(event.pollable()); + executePrefetch(queue, useV2); + assertEquals(1L, queue.getSubscriptionUncommittedEventCount()); + assertEquals(0, queue.getPrefetchedEventCount()); + + updateInterval(topicName, "300000"); + executePrefetch(queue, useV2); + assertEquals(0L, queue.getSubscriptionUncommittedEventCount()); + assertEquals(1, queue.getPrefetchedEventCount()); + assertEquals(1L, event.getNackCount()); + assertSame(event, poll(queue, useV2)); + } finally { + queue.cleanUp(); + SubscriptionAgent.topic().handleDropTopic(topicName, false); + CommonDescriptor.getInstance() + .getConfig() + .setSubscriptionRecycleUncommittedEventIntervalMs(originalInterval); + } + } + + private static void updateInterval(final String topicName, final String interval) { + assertNull( + SubscriptionAgent.topic() + .handleSingleTopicMetaChanges( + new TopicMeta( + topicName, + 1L, + Collections.singletonMap( + TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, interval)))); + } + + private static SubscriptionEvent poll( + final SubscriptionPrefetchingQueue queue, final boolean useV2) { + return useV2 + ? queue.pollV2("consumer", new PollTimer(System.currentTimeMillis(), 0L)) + : queue.poll("consumer"); + } + + private static void executePrefetch( + final SubscriptionPrefetchingQueue queue, final boolean useV2) { + if (useV2) { + queue.executePrefetchV2(); + } else { + queue.executePrefetch(); + } + } +} 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 c8bd2734e9174..1b7264c355897 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 @@ -23,7 +23,9 @@ import org.apache.iotdb.commons.consensus.DataRegionId; import org.apache.iotdb.commons.memory.AtomicLongMemoryBlock; import org.apache.iotdb.commons.memory.IMemoryBlock; +import org.apache.iotdb.commons.pipe.config.constant.SystemConstant; import org.apache.iotdb.commons.request.IConsensusRequest; +import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta; import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest; import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl; import org.apache.iotdb.consensus.iot.SubscriptionWalRetentionPolicy; @@ -81,6 +83,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.locks.ReentrantReadWriteLock; import java.util.function.Consumer; @@ -2865,6 +2868,111 @@ public void testWideTablePausedConsumerKeepsMaterializedMemoryBounded() throws E } } + @Test + public void testTopicRecycleIntervalOverridesGlobalAndAppliesUpdates() throws Exception { + final String topicName = "consensus_recycle_topic"; + final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); + final int originalRecycleInterval = + CommonDescriptor.getInstance() + .getConfig() + .getSubscriptionRecycleUncommittedEventIntervalMs(); + final long originalWatermarkInterval = + CommonDescriptor.getInstance().getConfig().getSubscriptionConsensusWatermarkIntervalMs(); + final File systemDir = temporaryFolder.newFolder("system-topic-recycle-interval"); + ConsensusPrefetchingQueue queue = null; + try { + CommonDescriptor.getInstance() + .getConfig() + .setSubscriptionRecycleUncommittedEventIntervalMs(600000); + CommonDescriptor.getInstance().getConfig().setSubscriptionConsensusWatermarkIntervalMs(0L); + assertNull( + SubscriptionAgent.topic() + .handleSingleTopicMetaChanges( + new TopicMeta( + topicName, + 1L, + java.util.Map.of( + SystemConstant.SQL_DIALECT_KEY, + SystemConstant.SQL_DIALECT_TABLE_VALUE, + TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, + "1200000")))); + final FakeConsensusReqReader reader = new FakeConsensusReqReader(); + final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); + when(serverImpl.getConsensusReqReader()).thenReturn(reader); + when(serverImpl.getWriterSafeFrontierTracker()).thenReturn(new WriterSafeFrontierTracker()); + final ConsensusLogToTabletConverter converter = mock(ConsensusLogToTabletConverter.class); + when(converter.convert(any())) + .thenReturn(Collections.singletonList(createTablet()), Collections.emptyList()); + when(converter.getDatabaseName()).thenReturn("db"); + when(converter.isTableModel()).thenReturn(true); + queue = + new ConsensusPrefetchingQueue( + "consumerGroup", + topicName, + TopicConstant.ORDER_MODE_LEADER_ONLY_VALUE, + new DataRegionId(1), + serverImpl, + new SubscriptionWalRetentionPolicy( + topicName, + SubscriptionWalRetentionPolicy.UNBOUNDED, + SubscriptionWalRetentionPolicy.UNBOUNDED), + converter, + newCommitManager(systemDir), + new RegionProgress(Collections.emptyMap()), + 1L, + 1L, + true); + queue.setSubscriptionMemoryManager( + new SubscriptionMemoryManager(createTablet().ramBytesUsed() * 2L)); + reader.currentSearchIndex = 2L; + assertTrue(pendingEntries(queue).offer(createRequest(1L))); + assertTrue(pendingEntries(queue).offer(createRequest(2L))); + assertNull(queue.poll("oldConsumer")); + queue.drivePrefetchOnce(); + final SubscriptionEvent event = queue.poll("oldConsumer"); + assertNotNull(event); + final Field timestampField = SubscriptionEvent.class.getDeclaredField("lastPolledTimestamp"); + timestampField.setAccessible(true); + ((AtomicLong) timestampField.get(event)).set(System.currentTimeMillis() - 900000L); + assertTrue(event.pollable()); + + final PrefetchRoundResult result = queue.drivePrefetchOnce(); + assertEquals(1L, queue.getSubscriptionUncommittedEventCount()); + assertEquals(0, queue.getPrefetchedEventCount()); + assertEquals(PrefetchRoundResult.Type.RESCHEDULE_LATER, result.getType()); + assertEquals(1200000L, result.getDelayMs()); + + assertNull( + SubscriptionAgent.topic() + .handleSingleTopicMetaChanges( + new TopicMeta( + topicName, + 1L, + java.util.Map.of( + SystemConstant.SQL_DIALECT_KEY, + SystemConstant.SQL_DIALECT_TABLE_VALUE, + TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, + "300000")))); + queue.drivePrefetchOnce(); + assertEquals(0L, queue.getSubscriptionUncommittedEventCount()); + assertEquals(1, queue.getPrefetchedEventCount()); + assertSame(event, queue.poll("newConsumer")); + assertTrue(queue.ack("newConsumer", event.getCommitContext())); + } finally { + if (queue != null) { + queue.close(); + } + SubscriptionAgent.topic().handleDropTopic(topicName, true); + CommonDescriptor.getInstance() + .getConfig() + .setSubscriptionRecycleUncommittedEventIntervalMs(originalRecycleInterval); + CommonDescriptor.getInstance() + .getConfig() + .setSubscriptionConsensusWatermarkIntervalMs(originalWatermarkInterval); + IoTDBDescriptor.getInstance().getConfig().setSystemDir(originalSystemDir); + } + } + @Test public void testLateAckAfterRecycleReleasesQueuedTabletMemory() throws Exception { final String originalSystemDir = IoTDBDescriptor.getInstance().getConfig().getSystemDir(); 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 b229f36d1fed5..d860d1e7b45d3 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 @@ -2160,6 +2160,16 @@ pipe_all_sinks_rate_limit_bytes_per_second=-1 ### Subscription Consensus Configuration #################### +# Time without a poll or in-flight lease renewal before an uncommitted subscription event is +# recycled for redelivery. Applies to all topic modes and both record and TsFile formats. +# A topic can override this with a positive long value in milliseconds, for example: +# CREATE TOPIC t WITH ('recycle-uncommitted-event-interval-ms'='1200000'); +# ALTER TOPIC t WITH ('recycle-uncommitted-event-interval-ms'='300000'); +# Topics without an override use this server configuration (default: 600 seconds). +# effectiveMode: restart +# Datatype: int, Unit: ms +subscription_recycle_uncommitted_event_interval_ms=600000 + # 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 From 20199527f7c4af533c202e7ee521c083b4f90b12 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 15 Oct 2026 11:32:46 +0800 Subject: [PATCH 2/2] [Subscription] Allow disabling uncommitted event recycling --- .../rpc/subscription/config/TopicConfig.java | 5 +- .../subscription/config/TopicConfigTest.java | 7 +++ .../confignode/i18n/ConfigNodeMessages.java | 4 +- .../confignode/i18n/ConfigNodeMessages.java | 4 +- .../subscription/SubscriptionInfo.java | 10 ++- .../SubscriptionInfoTopicValidationTest.java | 39 +++++++++--- .../consensus/ConsensusPrefetchingQueue.java | 5 +- .../subscription/event/SubscriptionEvent.java | 5 +- .../agent/SubscriptionTopicAgentTest.java | 13 +++- ...onPrefetchingQueueRecycleIntervalTest.java | 33 +++++++++- .../ConsensusPrefetchingQueueTest.java | 62 ++++++++++++++++++- .../conf/iotdb-system.properties.template | 5 +- 12 files changed, 162 insertions(+), 30 deletions(-) diff --git a/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java b/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java index ae371ccfd32de..bc4a5451cf34d 100644 --- a/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java +++ b/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java @@ -190,7 +190,10 @@ public static boolean isValidOrderMode(final String orderMode) { return ORDER_MODE_VALUE_SET.contains(normalizeOrderMode(orderMode)); } - /** Returns the topic's recycle timeout in milliseconds, or null to use the server config. */ + /** + * Returns the topic's recycle timeout in milliseconds, or null to use the server config. A zero + * or negative timeout disables automatic recycling of uncommitted events. + */ public Long getRecycleUncommittedEventIntervalMs() { return containsKeyIgnoreCase(TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY) ? Long.parseLong( diff --git a/iotdb-client/subscription/src/test/java/org/apache/iotdb/rpc/subscription/config/TopicConfigTest.java b/iotdb-client/subscription/src/test/java/org/apache/iotdb/rpc/subscription/config/TopicConfigTest.java index a1ae258c8cae7..0235feb4298dc 100644 --- a/iotdb-client/subscription/src/test/java/org/apache/iotdb/rpc/subscription/config/TopicConfigTest.java +++ b/iotdb-client/subscription/src/test/java/org/apache/iotdb/rpc/subscription/config/TopicConfigTest.java @@ -62,6 +62,13 @@ public void testRecycleUncommittedEventInterval() { attributes.put("RECYCLE-UNCOMMITTED-EVENT-INTERVAL-MS", "300000"); Assert.assertEquals( Long.valueOf(300000L), new TopicConfig(attributes).getRecycleUncommittedEventIntervalMs()); + + for (final long intervalMs : new long[] {0L, -1L, Long.MIN_VALUE}) { + attributes.put("RECYCLE-UNCOMMITTED-EVENT-INTERVAL-MS", " " + intervalMs + " "); + Assert.assertEquals( + Long.valueOf(intervalMs), + new TopicConfig(attributes).getRecycleUncommittedEventIntervalMs()); + } } @SuppressWarnings("deprecation") diff --git a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java index bffe196b708ba..60d672cc4901b 100644 --- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java +++ b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java @@ -21,8 +21,8 @@ public final class ConfigNodeMessages { public static final String - EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ILLEGAL_ARG_ARG_EXPECTED_A_POSITIVE_LONG_VALUE_IN_MILLISECONDS_8A5C0D8C = - "Failed to create or alter topic, illegal %s=%s, expected a positive long value in milliseconds"; + EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ILLEGAL_ARG_ARG_EXPECTED_A_LONG_VALUE_IN_MILLISECONDS_F79736B2 = + "Failed to create or alter topic, illegal %s=%s, expected a long value in milliseconds"; public static final String LOG_CONFIGREGION_UPDATE_CONFIGNODE_NODEID_ARG_CONSENSUSENDPOINT_ARG_IS_A_CONFIGREGION_FOLLOWER_NEW_LEADER_NODEID_ARG_STOPPING_LEADER_SERVICES_A1962CD2 = "[ConfigRegion Update] ConfigNode [nodeId: {}, consensusEndpoint: {}] is a ConfigRegion follower; new leader nodeId: {}. Stopping leader services."; diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java index b1910b2c3b91e..1b3926590cb82 100644 --- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java +++ b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java @@ -21,8 +21,8 @@ public final class ConfigNodeMessages { public static final String - EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ILLEGAL_ARG_ARG_EXPECTED_A_POSITIVE_LONG_VALUE_IN_MILLISECONDS_8A5C0D8C = - "创建或修改 topic 失败，%s=%s 无效，需指定以毫秒为单位的正 long 值"; + EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ILLEGAL_ARG_ARG_EXPECTED_A_LONG_VALUE_IN_MILLISECONDS_F79736B2 = + "创建或修改 topic 失败，%s=%s 无效，需指定以毫秒为单位的 long 值"; public static final String LOG_CONFIGREGION_UPDATE_CONFIGNODE_NODEID_ARG_CONSENSUSENDPOINT_ARG_IS_A_CONFIGREGION_FOLLOWER_NEW_LEADER_NODEID_ARG_STOPPING_LEADER_SERVICES_A1962CD2 = "[ConfigRegion Update] ConfigNode [nodeId: {}, consensusEndpoint: {}] 为 ConfigRegion follower；新 leader 的 nodeId: {}。开始停止 leader 服务。"; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java index 2796a0a1bf832..58fe0a10690c1 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfo.java @@ -592,17 +592,15 @@ private boolean isIncrementalTopicConfig(final TopicConfig topicConfig) { private void validateRecycleUncommittedEventInterval(final TopicConfig topicConfig) throws SubscriptionException { try { - final Long intervalMs = topicConfig.getRecycleUncommittedEventIntervalMs(); - if (Objects.isNull(intervalMs) || intervalMs > 0L) { - return; - } + topicConfig.getRecycleUncommittedEventIntervalMs(); + return; } catch (final NumberFormatException ignored) { - // Report malformed values together with non-positive timeouts below. + // Non-positive values disable automatic recycling; only malformed values are rejected. } final String exceptionMessage = String.format( ConfigNodeMessages - .EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ILLEGAL_ARG_ARG_EXPECTED_A_POSITIVE_LONG_VALUE_IN_MILLISECONDS_8A5C0D8C, + .EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ILLEGAL_ARG_ARG_EXPECTED_A_LONG_VALUE_IN_MILLISECONDS_F79736B2, TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, topicConfig.getAttribute().entrySet().stream() .filter( diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTopicValidationTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTopicValidationTest.java index 5b15b7af7f34d..f9e7e7a2f453a 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTopicValidationTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTopicValidationTest.java @@ -336,15 +336,36 @@ public void testAcceptRecycleUncommittedEventInterval() throws Exception { } } + @Test + public void testAcceptDisablingRecycleUncommittedEvents() throws Exception { + final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); + for (final String mode : + new String[] { + TopicConstant.MODE_INITIAL_VALUE, + TopicConstant.MODE_INCREMENTAL_VALUE, + TopicConstant.MODE_SNAPSHOT_VALUE + }) { + for (final String value : new String[] {"0", "-1", "-9223372036854775808"}) { + final Map attributes = newInitialTableTopicAttributes(); + attributes.put(TopicConstant.MODE_KEY, mode); + attributes.put(TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, value); + Assert.assertTrue( + subscriptionInfo.validateBeforeCreatingTopic( + new TCreateTopicReq("disabled_recycle_topic").setTopicAttributes(attributes))); + } + } + } + @Test public void testRejectInvalidRecycleUncommittedEventInterval() { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); for (final String value : - new String[] {"0", "-1", "not-a-number", "", " ", "1.5", "9223372036854775808", null}) { + new String[] { + "not-a-number", "", " ", "1.5", "9223372036854775808", "-9223372036854775809", null + }) { final Map attributes = newInitialTableTopicAttributes(); attributes.put(TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, value); - assertCreateRejected( - subscriptionInfo, attributes, "expected a positive long value in milliseconds"); + assertCreateRejected(subscriptionInfo, attributes, "expected a long value in milliseconds"); } } @@ -358,9 +379,11 @@ public void testAcceptAlteringRecycleUncommittedEventInterval() throws Exception attributes.put(TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, "1200000"); subscriptionInfo.createTopic(new CreateTopicPlan(new TopicMeta(mode, 1L, attributes))); - final Map updatedAttributes = new HashMap<>(attributes); - updatedAttributes.put(TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, "300000"); - subscriptionInfo.validateBeforeAlteringTopic(new TopicMeta(mode, 1L, updatedAttributes)); + for (final String value : new String[] {"300000", "0", "-1", "-9223372036854775808"}) { + final Map updatedAttributes = new HashMap<>(attributes); + updatedAttributes.put(TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, value); + subscriptionInfo.validateBeforeAlteringTopic(new TopicMeta(mode, 1L, updatedAttributes)); + } } } @@ -369,12 +392,12 @@ public void testRejectAlteringInvalidRecycleUncommittedEventInterval() { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); final Map attributes = newInitialTableTopicAttributes(); subscriptionInfo.createTopic(new CreateTopicPlan(new TopicMeta("table_topic", 1L, attributes))); - attributes.put(TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, "0"); + attributes.put(TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, "not-a-number"); try { subscriptionInfo.validateBeforeAlteringTopic(new TopicMeta("table_topic", 1L, attributes)); Assert.fail("Expected an invalid recycle interval to be rejected"); } catch (final SubscriptionException e) { - Assert.assertTrue(e.getMessage().contains("expected a positive long value in milliseconds")); + Assert.assertTrue(e.getMessage().contains("expected a long value in milliseconds")); } } 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 90e6df8ec1883..9ee6079dabb25 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 @@ -1931,7 +1931,10 @@ private PrefetchRoundResult computeIdleRoundResult( } if (!inFlightEvents.isEmpty()) { - delayMs = Math.min(delayMs, getRecycleUncommittedEventIntervalMs()); + final long recycleIntervalMs = getRecycleUncommittedEventIntervalMs(); + if (recycleIntervalMs > 0L) { + delayMs = Math.min(delayMs, recycleIntervalMs); + } } return delayMs == Long.MAX_VALUE diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/SubscriptionEvent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/SubscriptionEvent.java index d055c2855a113..031e68780a244 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/SubscriptionEvent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/event/SubscriptionEvent.java @@ -263,8 +263,9 @@ public boolean eagerlyPollable() { private boolean canRecycle(final long recycleUncommittedEventIntervalMs) { // Recycle events that may not be able to be committed, i.e., those that have been polled but // not committed within a certain period of time. - return System.currentTimeMillis() - lastPolledTimestamp.get() - > recycleUncommittedEventIntervalMs; + return recycleUncommittedEventIntervalMs > 0L + && System.currentTimeMillis() - lastPolledTimestamp.get() + > recycleUncommittedEventIntervalMs; } public void nack() { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgentTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgentTest.java index 1a658fdf1da83..1b09a53f55907 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgentTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionTopicAgentTest.java @@ -48,6 +48,14 @@ public void testRecycleIntervalUpdateAndFallback() { Assert.assertEquals( 1200000L, agent.getTopicRecycleUncommittedEventIntervalMs("recycle_topic", false)); + for (final long intervalMs : new long[] {0L, -1L, Long.MIN_VALUE}) { + Assert.assertNull( + agent.handleSingleTopicMetaChanges( + createRecycleTopicMeta(false, Long.toString(intervalMs)))); + Assert.assertEquals( + intervalMs, agent.getTopicRecycleUncommittedEventIntervalMs("recycle_topic", false)); + } + Assert.assertNull(agent.handleSingleTopicMetaChanges(createRecycleTopicMeta(false, "300000"))); Assert.assertEquals( 300000L, agent.getTopicRecycleUncommittedEventIntervalMs("recycle_topic", false)); @@ -83,13 +91,12 @@ public void testRecycleIntervalIsolationAndDrop() { @Test public void testRecycleIntervalSnapshotPreservesCurrentOverride() { final SubscriptionTopicAgent agent = new SubscriptionTopicAgent(); - final TopicMeta topicMeta = spy(createRecycleTopicMeta(false, "1200000")); + final TopicMeta topicMeta = spy(createRecycleTopicMeta(false, "0")); Assert.assertNull(agent.handleSingleTopicMetaChanges(topicMeta)); doAnswer( invocation -> { Assert.assertEquals( - 1200000L, - agent.getTopicRecycleUncommittedEventIntervalMs("recycle_topic", false)); + 0L, agent.getTopicRecycleUncommittedEventIntervalMs("recycle_topic", false)); return invocation.callRealMethod(); }) .when(topicMeta) diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueueRecycleIntervalTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueueRecycleIntervalTest.java index abebe01d51709..3aa26ed54b106 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueueRecycleIntervalTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueueRecycleIntervalTest.java @@ -36,6 +36,7 @@ import java.util.concurrent.atomic.AtomicLong; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertTrue; @@ -76,6 +77,7 @@ private static void verifyRecycleInterval(final boolean useV2) throws Exception SubscriptionPollResponseType.TERMINATION.getType(), new TerminationPayload(), new SubscriptionCommitContext(1, 1, topicName, "recycle_group", 1)); + assertTrue(event.pollable(0L)); queue.prefetchEvent(event); assertSame(event, poll(queue, useV2)); @@ -87,12 +89,37 @@ private static void verifyRecycleInterval(final boolean useV2) throws Exception assertEquals(1L, queue.getSubscriptionUncommittedEventCount()); assertEquals(0, queue.getPrefetchedEventCount()); + for (final long intervalMs : new long[] {0L, -1L, Long.MIN_VALUE}) { + updateInterval(topicName, Long.toString(intervalMs)); + executePrefetch(queue, useV2); + assertEquals(1L, queue.getSubscriptionUncommittedEventCount()); + assertEquals(0, queue.getPrefetchedEventCount()); + assertEquals(0L, event.getNackCount()); + } + + updateInterval(topicName, null); + for (final int intervalMs : new int[] {0, -1, Integer.MIN_VALUE}) { + CommonDescriptor.getInstance() + .getConfig() + .setSubscriptionRecycleUncommittedEventIntervalMs(intervalMs); + assertFalse(event.pollable()); + executePrefetch(queue, useV2); + assertEquals(1L, queue.getSubscriptionUncommittedEventCount()); + assertEquals(0, queue.getPrefetchedEventCount()); + } + updateInterval(topicName, "300000"); executePrefetch(queue, useV2); assertEquals(0L, queue.getSubscriptionUncommittedEventCount()); assertEquals(1, queue.getPrefetchedEventCount()); assertEquals(1L, event.getNackCount()); assertSame(event, poll(queue, useV2)); + + updateInterval(topicName, "0"); + assertTrue(queue.nack("consumer", event.getCommitContext())); + executePrefetch(queue, useV2); + assertSame(event, poll(queue, useV2)); + assertTrue(queue.ack("consumer", event.getCommitContext())); } finally { queue.cleanUp(); SubscriptionAgent.topic().handleDropTopic(topicName, false); @@ -109,8 +136,10 @@ private static void updateInterval(final String topicName, final String interval new TopicMeta( topicName, 1L, - Collections.singletonMap( - TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, interval)))); + interval == null + ? Collections.emptyMap() + : Collections.singletonMap( + TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, interval)))); } private static SubscriptionEvent poll( 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 1b7264c355897..7cfd6abaf45cf 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 @@ -2942,6 +2942,43 @@ public void testTopicRecycleIntervalOverridesGlobalAndAppliesUpdates() throws Ex assertEquals(PrefetchRoundResult.Type.RESCHEDULE_LATER, result.getType()); assertEquals(1200000L, result.getDelayMs()); + for (final long intervalMs : new long[] {0L, -1L, Long.MIN_VALUE}) { + assertNull( + SubscriptionAgent.topic() + .handleSingleTopicMetaChanges( + new TopicMeta( + topicName, + 1L, + java.util.Map.of( + SystemConstant.SQL_DIALECT_KEY, + SystemConstant.SQL_DIALECT_TABLE_VALUE, + TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, + Long.toString(intervalMs))))); + assertEquals(PrefetchRoundResult.Type.DORMANT, queue.drivePrefetchOnce().getType()); + assertEquals(1L, queue.getSubscriptionUncommittedEventCount()); + assertEquals(0, queue.getPrefetchedEventCount()); + assertEquals(0L, event.getNackCount()); + } + + assertNull( + SubscriptionAgent.topic() + .handleSingleTopicMetaChanges( + new TopicMeta( + topicName, + 1L, + java.util.Map.of( + SystemConstant.SQL_DIALECT_KEY, + SystemConstant.SQL_DIALECT_TABLE_VALUE)))); + for (final int intervalMs : new int[] {0, -1, Integer.MIN_VALUE}) { + CommonDescriptor.getInstance() + .getConfig() + .setSubscriptionRecycleUncommittedEventIntervalMs(intervalMs); + assertFalse(event.pollable()); + assertEquals(PrefetchRoundResult.Type.DORMANT, queue.drivePrefetchOnce().getType()); + assertEquals(1L, queue.getSubscriptionUncommittedEventCount()); + assertEquals(0, queue.getPrefetchedEventCount()); + } + assertNull( SubscriptionAgent.topic() .handleSingleTopicMetaChanges( @@ -2957,6 +2994,21 @@ public void testTopicRecycleIntervalOverridesGlobalAndAppliesUpdates() throws Ex assertEquals(0L, queue.getSubscriptionUncommittedEventCount()); assertEquals(1, queue.getPrefetchedEventCount()); assertSame(event, queue.poll("newConsumer")); + assertEquals(1L, event.getNackCount()); + assertNull( + SubscriptionAgent.topic() + .handleSingleTopicMetaChanges( + new TopicMeta( + topicName, + 1L, + java.util.Map.of( + SystemConstant.SQL_DIALECT_KEY, + SystemConstant.SQL_DIALECT_TABLE_VALUE, + TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, + "0")))); + assertTrue(queue.nack("newConsumer", event.getCommitContext())); + queue.drivePrefetchOnce(); + assertSame(event, queue.poll("newConsumer")); assertTrue(queue.ack("newConsumer", event.getCommitContext())); } finally { if (queue != null) { @@ -2985,7 +3037,7 @@ public void testLateAckAfterRecycleReleasesQueuedTabletMemory() throws Exception try { CommonDescriptor.getInstance() .getConfig() - .setSubscriptionRecycleUncommittedEventIntervalMs(-1); + .setSubscriptionRecycleUncommittedEventIntervalMs(1); final DataRegionId regionId = new DataRegionId(1); final FakeConsensusReqReader reader = new FakeConsensusReqReader(); final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); @@ -3029,6 +3081,9 @@ public void testLateAckAfterRecycleReleasesQueuedTabletMemory() throws Exception assertTrue(queue.getRetainedTabletBytes() > 0L); assertEquals(1L, queue.getSubscriptionUncommittedEventCount()); + final Field timestampField = SubscriptionEvent.class.getDeclaredField("lastPolledTimestamp"); + timestampField.setAccessible(true); + ((AtomicLong) timestampField.get(event)).set(System.currentTimeMillis() - 900000L); queue.drivePrefetchOnce(); assertEquals(0L, queue.getSubscriptionUncommittedEventCount()); assertEquals(1, queue.getPrefetchedEventCount()); @@ -3060,7 +3115,7 @@ public void testLateAckDoesNotStealRecycledEventFromNewConsumer() throws Excepti try { CommonDescriptor.getInstance() .getConfig() - .setSubscriptionRecycleUncommittedEventIntervalMs(-1); + .setSubscriptionRecycleUncommittedEventIntervalMs(1); final DataRegionId regionId = new DataRegionId(1); final FakeConsensusReqReader reader = new FakeConsensusReqReader(); final IoTConsensusServerImpl serverImpl = mock(IoTConsensusServerImpl.class); @@ -3104,6 +3159,9 @@ public void testLateAckDoesNotStealRecycledEventFromNewConsumer() throws Excepti final long retainedBytes = queue.getRetainedTabletBytes(); assertTrue(retainedBytes > 0L); + final Field timestampField = SubscriptionEvent.class.getDeclaredField("lastPolledTimestamp"); + timestampField.setAccessible(true); + ((AtomicLong) timestampField.get(event)).set(System.currentTimeMillis() - 900000L); queue.drivePrefetchOnce(); assertEquals(0L, queue.getSubscriptionUncommittedEventCount()); assertEquals(1, queue.getPrefetchedEventCount()); 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 d860d1e7b45d3..0a1de7d02f155 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 @@ -2162,9 +2162,12 @@ pipe_all_sinks_rate_limit_bytes_per_second=-1 # Time without a poll or in-flight lease renewal before an uncommitted subscription event is # recycled for redelivery. Applies to all topic modes and both record and TsFile formats. -# A topic can override this with a positive long value in milliseconds, for example: +# A zero or negative value disables automatic recycling. A topic can override this with a long value +# in milliseconds; zero or negative topic values also disable automatic recycling, for example: # CREATE TOPIC t WITH ('recycle-uncommitted-event-interval-ms'='1200000'); # ALTER TOPIC t WITH ('recycle-uncommitted-event-interval-ms'='300000'); +# ALTER TOPIC t WITH ('recycle-uncommitted-event-interval-ms'='0'); +# Disabling automatic recycling still allows explicit NACK redelivery and ACK commits. # Topics without an override use this server configuration (default: 600 seconds). # effectiveMode: restart # Datatype: int, Unit: ms