[Subscription] Support per-topic uncommitted event recycle timeouts

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

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

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<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() {

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.";

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 服务。";

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)

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<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();

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<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 {

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<Pair<String, Boolean>, 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<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 TPushTopicMetaRespExceptionMessage handleTopicMetaChanges(
       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 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<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 @@ 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) {

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<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);