This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new e3bbab5e7e8 [Subscription] Support per-topic uncommitted event recycle
timeouts (#18876)
e3bbab5e7e8 is described below
commit e3bbab5e7e851b7411a5655b67996c8fcae631cf
Author: Caideyipi <[email protected]>
AuthorDate: Thu Oct 15 14:12:52 2026 +0800
[Subscription] Support per-topic uncommitted event recycle timeouts (#18876)
* [Subscription] Support per-topic uncommitted event recycle timeouts
* [Subscription] Allow disabling uncommitted event recycling
---
.../iotdb/rpc/subscription/config/TopicConfig.java | 11 ++
.../rpc/subscription/config/TopicConstant.java | 4 +
.../rpc/subscription/config/TopicConfigTest.java | 22 +++
.../iotdb/confignode/i18n/ConfigNodeMessages.java | 3 +
.../iotdb/confignode/i18n/ConfigNodeMessages.java | 3 +
.../persistence/subscription/SubscriptionInfo.java | 28 ++++
.../SubscriptionInfoTopicValidationTest.java | 106 +++++++++++++
.../agent/SubscriptionBrokerAgent.java | 10 ++
.../subscription/agent/SubscriptionTopicAgent.java | 47 ++++++
.../broker/ConsensusSubscriptionBroker.java | 8 +
.../broker/SubscriptionPrefetchingQueue.java | 13 +-
.../consensus/ConsensusPrefetchingQueue.java | 23 ++-
.../db/subscription/event/SubscriptionEvent.java | 14 +-
.../agent/SubscriptionTopicAgentTest.java | 116 ++++++++++++++
...riptionPrefetchingQueueRecycleIntervalTest.java | 160 +++++++++++++++++++
.../consensus/ConsensusPrefetchingQueueTest.java | 170 ++++++++++++++++++++-
.../conf/iotdb-system.properties.template | 13 ++
17 files changed, 736 insertions(+), 15 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 51b87f910e2..bc4a5451cf3 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 class TopicConfig extends PipeParameters {
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 960e3bab524..8a8af71d738 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 3b32d407eba..0235feb4298 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 class TopicConfigTest {
Assert.assertTrue(topicConfigWithMode(" INCREMENTAL
").isIncrementalMode());
}
+ @Test
+ public void testRecycleUncommittedEventInterval() {
+ Assert.assertNull(new
TopicConfig().getRecycleUncommittedEventIntervalMs());
+
+ final Map<String, String> 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 fb73cc93376..60d672cc490 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 948f4a64765..1b3926590cb 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 9fab5d1f536..58fe0a10690 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 @@ public class SubscriptionInfo implements SnapshotProcessor {
validateColumnFilter(topicConfig);
validateIncrementalTopicRetentionConfig(topicConfig);
+ validateRecycleUncommittedEventInterval(topicConfig);
final Long ownerLeaseDurationMs =
topicConfig.getLong(TopicConstant.OWNER_LEASE_DURATION_MS_KEY);
@@ -586,6 +589,31 @@ public class SubscriptionInfo implements SnapshotProcessor
{
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 da9c356190c..f9e7e7a2f45 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 class SubscriptionInfoTopicValidationTest {
assertCreateRejected(subscriptionInfo, attributes, "expected a long
value");
}
+ @Test
+ public void testAcceptRecycleUncommittedEventInterval() throws Exception {
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+
+ final Map<String, String> initialAttributes =
newInitialTableTopicAttributes();
+ initialAttributes.put(
+ TopicConstant.RECYCLE_UNCOMMITTED_EVENT_INTERVAL_MS_KEY, "
9223372036854775807 ");
+ Assert.assertTrue(
+ subscriptionInfo.validateBeforeCreatingTopic(
+ new
TCreateTopicReq("initial_topic").setTopicAttributes(initialAttributes)));
+
+ final Map<String, String> 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<String, String> 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<String, String> 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<String, String> 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<String, String> 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<String, String> 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<String, String> 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<String, String> 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 1c429cd97dc..aafcf903dc4 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 class SubscriptionBrokerAgent {
}
}
+ public void wakeUpConsensusPrefetchingQueues(final String topicName, final
boolean isTableModel) {
+ for (final List<ISubscriptionBroker> 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 f4cb98d823e..f68e60f353f 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.config.constant.SystemConstant;
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.ConsumerConfig;
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<Pair<String, Boolean>, Long> topicRecycleIntervalsMs =
+ new ConcurrentHashMap<>();
+
public SubscriptionTopicAgent() {
this.topicMetaKeeper = new TopicMetaKeeper();
}
@@ -109,8 +119,18 @@ public class SubscriptionTopicAgent {
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<String, Boolean> 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 class SubscriptionTopicAgent {
final List<TopicMeta> topicMetasFromCoordinator) {
acquireWriteLock();
try {
+ final Set<Pair<String, Boolean>> 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 class SubscriptionTopicAgent {
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<String, Boolean> 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 @@ public class SubscriptionTopicAgent {
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 cd7abf043cf..60b268504a2 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 class ConsensusSubscriptionBroker implements
ISubscriptionBroker {
closeAndRemoveConsensusPrefetchingQueues(topicName, true,
retainProgressAfterUnsubscribe);
}
+ public void wakeUpPrefetchingQueues(final String topicName, final boolean
isTableModel) {
+ final List<ConsensusPrefetchingQueue> 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 716e7d90f15..2756f4fa4a9 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 abstract class 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 abstract class SubscriptionPrefetchingQueue {
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 @@ public abstract class SubscriptionPrefetchingQueue {
continue;
}
- if (!event.pollable()) {
+ if (!event.pollable(getRecycleUncommittedEventIntervalMs())) {
LOGGER.warn(
DataNodePipeMessages
.PIPE_LOG_SUBSCRIPTION_SUBSCRIPTIONPREFETCHINGQUEUE_POLL_NON_POLLABLE_644D5D6B,
@@ -373,7 +380,7 @@ public abstract class SubscriptionPrefetchingQueue {
continue;
}
- if (!event.pollable()) {
+ if (!event.pollable(getRecycleUncommittedEventIntervalMs())) {
LOGGER.warn(
DataNodePipeMessages
.PIPE_LOG_SUBSCRIPTION_SUBSCRIPTIONPREFETCHINGQUEUE_POLL_NON_POLLABLE_644D5D6B,
@@ -1232,7 +1239,7 @@ public abstract class SubscriptionPrefetchingQueue {
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 1dd733b4cae..4e71f2e2fe9 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 @@ public class ConsensusPrefetchingQueue {
}
}
+ 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 @@ public class ConsensusPrefetchingQueue {
continue;
}
- if (!event.pollable()) {
+ if (!event.pollable(getRecycleUncommittedEventIntervalMs())) {
LOGGER.warn(
DataNodePipeMessages
.PIPE_LOG_CONSENSUSPREFETCHINGQUEUE_POLL_NON_POLLABLE_EVENT_BROKEN_E9551325,
@@ -1939,10 +1950,10 @@ public class ConsensusPrefetchingQueue {
}
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 @@ public class ConsensusPrefetchingQueue {
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 2e58a68411d..031e68780a2 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 class SubscriptionEvent implements
Comparable<SubscriptionEvent> {
* 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 class SubscriptionEvent implements
Comparable<SubscriptionEvent> {
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 c31535b5812..1b09a53f559 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.Assert;
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<String, String> 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 00000000000..3aa26ed54b1
--- /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 32f8e5737fd..cc30193647a 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.conf.CommonDescriptor;
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.Future;
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 class ConsensusPrefetchingQueueTest {
}
}
+ @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 class ConsensusPrefetchingQueueTest {
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 class ConsensusPrefetchingQueueTest {
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 class ConsensusPrefetchingQueueTest {
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 class ConsensusPrefetchingQueueTest {
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 b229f36d1fe..0a1de7d02f1 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
