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 51b87f910e2d..bc4a5451cf34 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,17 @@ 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. 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( + 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 960e3bab524b..8a8af71d7388 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 3b32d407eba9..0235feb4298d 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,28 @@ 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()); + + 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") @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 fb73cc933763..60d672cc4901 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_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 948f4a647651..1b3926590cb8 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_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 9fab5d1f536b..58fe0a10690c 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,31 @@ private boolean isIncrementalTopicConfig(final TopicConfig topicConfig) { return topicConfig.isIncrementalMode(); } + private void validateRecycleUncommittedEventInterval(final TopicConfig topicConfig) + throws SubscriptionException { + try { + topicConfig.getRecycleUncommittedEventIntervalMs(); + return; + } catch (final NumberFormatException ignored) { + // 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_LONG_VALUE_IN_MILLISECONDS_F79736B2, + 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 da9c356190c6..f9e7e7a2f453 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,112 @@ 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 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[] { + "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 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))); + + 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)); + } + } + } + + @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, "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 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 1c429cd97dca..aafcf903dc43 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 f4cb98d823ed..f68e60f353f8 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 cd7abf043cf2..60b268504a2f 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 716e7d90f152..2756f4fa4a97 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 1dd733b4cae0..4e71f2e2fe97 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 @@ -752,6 +752,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 (!hasReachedPrefetchLimit()) { requestPrefetch(); @@ -1506,7 +1517,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, @@ -1939,10 +1950,10 @@ private PrefetchRoundResult computeIdleRoundResult( } if (!inFlightEvents.isEmpty()) { - delayMs = - Math.min( - delayMs, - SubscriptionConfig.getInstance().getSubscriptionRecycleUncommittedEventIntervalMs()); + final long recycleIntervalMs = getRecycleUncommittedEventIntervalMs(); + if (recycleIntervalMs > 0L) { + delayMs = Math.min(delayMs, recycleIntervalMs); + } } return delayMs == Long.MAX_VALUE @@ -3430,7 +3441,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 2e58a68411da..031e68780a24 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,12 @@ 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(); + 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 c31535b5812f..1b09a53f5590 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,124 @@ 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)); + + 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)); + + 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, "0")); + Assert.assertNull(agent.handleSingleTopicMetaChanges(topicMeta)); + doAnswer( + invocation -> { + Assert.assertEquals( + 0L, 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 000000000000..3aa26ed54b10 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/broker/SubscriptionPrefetchingQueueRecycleIntervalTest.java @@ -0,0 +1,160 @@ +/* + * 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.assertFalse; +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)); + assertTrue(event.pollable(0L)); + 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()); + + 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); + CommonDescriptor.getInstance() + .getConfig() + .setSubscriptionRecycleUncommittedEventIntervalMs(originalInterval); + } + } + + private static void updateInterval(final String topicName, final String interval) { + assertNull( + SubscriptionAgent.topic() + .handleSingleTopicMetaChanges( + new TopicMeta( + topicName, + 1L, + interval == null + ? Collections.emptyMap() + : 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 32f8e5737fd4..cc30193647af 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; @@ -2880,6 +2883,163 @@ 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()); + + 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( + 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")); + 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) { + 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(); @@ -2892,7 +3052,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); @@ -2936,6 +3096,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()); @@ -2967,7 +3130,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); @@ -3011,6 +3174,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 b229f36d1fed..0a1de7d02f15 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,19 @@ 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 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 +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