clientHandler =
+ new DataNodeAsyncRequestContext<>(
+ CnToDnAsyncRequestType.SUBSCRIPTION_PUSH_OWNER_LEASE, request, dataNodeLocationMap);
+ CnToDnInternalServiceAsyncRequestManager.getInstance().sendAsyncRequestWithRetry(clientHandler);
+ }
+
public TSStatus dropTopic(TDropTopicReq req) {
final String topicName = req.getTopicName();
final boolean isSetIfExistsCondition =
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionOwnerLeaseSyncer.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionOwnerLeaseSyncer.java
new file mode 100644
index 0000000000000..bbc22a35c8d84
--- /dev/null
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionOwnerLeaseSyncer.java
@@ -0,0 +1,97 @@
+/*
+ * 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.confignode.manager.subscription;
+
+import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
+import org.apache.iotdb.commons.concurrent.threadpool.ScheduledExecutorUtil;
+import org.apache.iotdb.confignode.manager.ConfigManager;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Dedicated, fine-grained subscription owner-lease heartbeat, independent from the node heartbeat.
+ *
+ * Each tick pushes relative-duration owner leases to all DataNodes (see {@link
+ * SubscriptionCoordinator#pushTopicOwnerLeasesToDataNodes()}). A DataNode that stops receiving
+ * pushes (e.g. partitioned from this ConfigNode) lets its local lease expire and fences the owner
+ * (fail-closed). The interval must stay well below the configured lease duration so transient
+ * misses do not spuriously fence a healthy owner.
+ */
+public class SubscriptionOwnerLeaseSyncer {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(SubscriptionOwnerLeaseSyncer.class);
+
+ // Note: candidate for promotion to SubscriptionConfig; invariant must hold: lease-duration >>
+ // interval * tolerated-misses.
+ private static final long OWNER_LEASE_HEARTBEAT_INTERVAL_SECONDS = 5;
+
+ /** The owner-lease heartbeat interval {@code H}, used by the transfer admission-wait bound. */
+ public static long getHeartbeatIntervalMs() {
+ return OWNER_LEASE_HEARTBEAT_INTERVAL_SECONDS * 1000L;
+ }
+
+ private static final ScheduledExecutorService LEASE_EXECUTOR =
+ IoTDBThreadPoolFactory.newSingleThreadScheduledExecutor("SubscriptionOwnerLeaseSyncer");
+
+ private final ConfigManager configManager;
+
+ private Future> leaseSyncFuture;
+
+ SubscriptionOwnerLeaseSyncer(final ConfigManager configManager) {
+ this.configManager = configManager;
+ }
+
+ public synchronized void start() {
+ if (leaseSyncFuture == null) {
+ leaseSyncFuture =
+ ScheduledExecutorUtil.safelyScheduleWithFixedDelay(
+ LEASE_EXECUTOR,
+ this::sync,
+ OWNER_LEASE_HEARTBEAT_INTERVAL_SECONDS,
+ OWNER_LEASE_HEARTBEAT_INTERVAL_SECONDS,
+ TimeUnit.SECONDS);
+ LOGGER.info("SubscriptionOwnerLeaseSyncer is started successfully.");
+ }
+ }
+
+ private void sync() {
+ try {
+ configManager
+ .getSubscriptionManager()
+ .getSubscriptionCoordinator()
+ .pushTopicOwnerLeasesToDataNodes();
+ } catch (final Exception e) {
+ LOGGER.warn("Failed to push subscription topic owner leases to DataNodes.", e);
+ }
+ }
+
+ public synchronized void stop() {
+ if (leaseSyncFuture != null) {
+ leaseSyncFuture.cancel(false);
+ leaseSyncFuture = null;
+ LOGGER.info("SubscriptionOwnerLeaseSyncer is stopped successfully.");
+ }
+ }
+}
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 ee0161ba085e1..a013b3d7f920e 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
@@ -21,6 +21,7 @@
import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.snapshot.SnapshotProcessor;
+import org.apache.iotdb.commons.subscription.config.SubscriptionConfig;
import org.apache.iotdb.commons.subscription.meta.consumer.CommitProgressKeeper;
import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerGroupMeta;
import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerGroupMetaKeeper;
@@ -40,6 +41,7 @@
import org.apache.iotdb.confignode.consensus.response.subscription.SubscriptionTableResp;
import org.apache.iotdb.confignode.consensus.response.subscription.TopicTableResp;
import org.apache.iotdb.confignode.i18n.ConfigNodeMessages;
+import org.apache.iotdb.confignode.i18n.ManagerMessages;
import org.apache.iotdb.confignode.rpc.thrift.TCloseConsumerReq;
import org.apache.iotdb.confignode.rpc.thrift.TCreateConsumerReq;
import org.apache.iotdb.confignode.rpc.thrift.TCreateTopicReq;
@@ -47,6 +49,7 @@
import org.apache.iotdb.confignode.rpc.thrift.TUnsubscribeReq;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.consensus.common.DataSet;
+import org.apache.iotdb.mpp.rpc.thrift.TTopicOwnerLeaseEntry;
import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.rpc.subscription.config.TopicConfig;
import org.apache.iotdb.rpc.subscription.config.TopicConstant;
@@ -332,6 +335,20 @@ private void validateTopicConfig(final TopicConfig topicConfig) throws Subscript
validateConsensusTableColumnPattern(topicConfig);
validateConsensusTopicRetentionConfig(topicConfig);
+
+ final Long ownerLeaseDurationMs =
+ topicConfig.getLong(TopicConstant.OWNER_LEASE_DURATION_MS_KEY);
+ final long ownerLeaseDurationMsMin =
+ SubscriptionConfig.getInstance().getSubscriptionOwnerLeaseDurationMsMin();
+ if (Objects.nonNull(ownerLeaseDurationMs) && ownerLeaseDurationMs < ownerLeaseDurationMsMin) {
+ final String exceptionMessage =
+ String.format(
+ ManagerMessages.OWNER_LEASE_DURATION_BELOW_MIN,
+ ownerLeaseDurationMs,
+ ownerLeaseDurationMsMin);
+ LOGGER.warn(exceptionMessage);
+ throw new SubscriptionException(exceptionMessage);
+ }
}
private void validateConsensusProtocolSupport(final TopicConfig topicConfig)
@@ -552,6 +569,49 @@ public TopicMeta deepCopyTopicMeta(String topicName) {
}
}
+ public TopicMeta deepCopyTopicMetaWithUpdatedAttributes(
+ String topicName, Map updatedAttributes) {
+ acquireReadLock();
+ try {
+ return topicMetaKeeper.containsTopicMeta(topicName)
+ ? topicMetaKeeper.getTopicMeta(topicName).deepCopyWithUpdatedAttributes(updatedAttributes)
+ : null;
+ } finally {
+ releaseReadLock();
+ }
+ }
+
+ /**
+ * Collect owner-lease entries to push via the dedicated subscription owner heartbeat. Each entry
+ * carries a relative remaining duration (the configured lease duration); the DataNode derives the
+ * local expire time on its own clock. Topics undergoing an owner transfer ({@code
+ * blockedTopicNames}) are skipped so their lease drains. This is read-only on ConfigNode: no
+ * absolute expire timestamp is ever stored or compared across nodes.
+ */
+ public List collectTopicOwnerLeaseEntries(
+ final Set blockedTopicNames) {
+ acquireReadLock();
+ try {
+ final List entries = new ArrayList<>();
+ for (final TopicMeta topicMeta : topicMetaKeeper.getAllTopicMeta()) {
+ if (!topicMeta.isOwnerFencingEnabled()
+ || Objects.isNull(topicMeta.getOwnerLeaseDurationMs())
+ || blockedTopicNames.contains(topicMeta.getTopicName())) {
+ continue;
+ }
+ entries.add(
+ new TTopicOwnerLeaseEntry(
+ topicMeta.getTopicName(),
+ topicMeta.getOwnerId(),
+ topicMeta.getOwnerEpoch(),
+ topicMeta.getOwnerLeaseDurationMs()));
+ }
+ return entries;
+ } finally {
+ releaseReadLock();
+ }
+ }
+
public DataSet showTopics() {
acquireReadLock();
try {
@@ -584,6 +644,14 @@ public TSStatus alterTopic(AlterTopicPlan plan) {
}
private TSStatus alterTopicInternal(final AlterTopicPlan plan) {
+ try {
+ TopicMeta.validateOwnerProgression(
+ topicMetaKeeper.getTopicMeta(plan.getTopicMeta().getTopicName()), plan.getTopicMeta());
+ } catch (final IllegalArgumentException e) {
+ return new TSStatus(TSStatusCode.SUBSCRIPTION_OWNER_EPOCH_CONFLICT.getStatusCode())
+ .setMessage(e.getMessage());
+ }
+
topicMetaKeeper.removeTopicMeta(plan.getTopicMeta().getTopicName());
topicMetaKeeper.addTopicMeta(plan.getTopicMeta().getTopicName(), plan.getTopicMeta());
return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
index efd3f8abe8018..5b26a6f7d6dc7 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/thrift/ConfigNodeRPCServiceProcessor.java
@@ -102,6 +102,7 @@
import org.apache.iotdb.confignode.rpc.thrift.TAlterPipeReq;
import org.apache.iotdb.confignode.rpc.thrift.TAlterSchemaTemplateReq;
import org.apache.iotdb.confignode.rpc.thrift.TAlterTimeSeriesReq;
+import org.apache.iotdb.confignode.rpc.thrift.TAlterTopicReq;
import org.apache.iotdb.confignode.rpc.thrift.TAuthizedPatternTreeResp;
import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerRelationalReq;
import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerReq;
@@ -1275,6 +1276,11 @@ public TSStatus createTopic(TCreateTopicReq req) {
return configManager.createTopic(req);
}
+ @Override
+ public TSStatus alterTopic(TAlterTopicReq req) {
+ return configManager.alterTopic(req);
+ }
+
@Override
public TSStatus dropTopic(String topicName) {
return configManager.dropTopic(
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTest.java
new file mode 100644
index 0000000000000..9b86f1d402af0
--- /dev/null
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/persistence/subscription/SubscriptionInfoTest.java
@@ -0,0 +1,194 @@
+/*
+ * 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.confignode.persistence.subscription;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta;
+import org.apache.iotdb.confignode.consensus.request.write.subscription.topic.AlterTopicPlan;
+import org.apache.iotdb.confignode.consensus.request.write.subscription.topic.CreateTopicPlan;
+import org.apache.iotdb.confignode.consensus.response.subscription.TopicTableResp;
+import org.apache.iotdb.confignode.rpc.thrift.TShowTopicInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TShowTopicResp;
+import org.apache.iotdb.mpp.rpc.thrift.TTopicOwnerLeaseEntry;
+import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.rpc.subscription.config.TopicConstant;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+
+public class SubscriptionInfoTest {
+
+ @Test
+ public void testAlterTopicRejectsOwnerEpochRollback() {
+ final String topicName = "topic-" + UUID.randomUUID();
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+
+ final TopicMeta initialTopicMeta = createTopicMeta(topicName, "owner1", 5L);
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ subscriptionInfo.createTopic(new CreateTopicPlan(initialTopicMeta)).getCode());
+
+ final TopicMeta transferredTopicMeta = initialTopicMeta.deepCopy();
+ transferredTopicMeta.transferOwner("owner2", 6L);
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ subscriptionInfo.alterTopic(new AlterTopicPlan(transferredTopicMeta)).getCode());
+
+ final TSStatus rollbackStatus =
+ subscriptionInfo.alterTopic(new AlterTopicPlan(createTopicMeta(topicName, "owner1", 5L)));
+
+ Assert.assertEquals(
+ TSStatusCode.SUBSCRIPTION_OWNER_EPOCH_CONFLICT.getStatusCode(), rollbackStatus.getCode());
+ Assert.assertEquals("owner2", subscriptionInfo.getTopicMeta(topicName).getOwnerId());
+ Assert.assertEquals(6L, subscriptionInfo.getTopicMeta(topicName).getOwnerEpoch());
+ }
+
+ @Test
+ public void testAlterTopicTransfersOwnerWithUpdatedAttributes() {
+ final String topicName = "topic-" + UUID.randomUUID();
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+
+ final TopicMeta initialTopicMeta = createTopicMeta(topicName, "owner1", 5L);
+ initialTopicMeta.getConfig().getAttribute().put(TopicConstant.PATH_KEY, "root.sg.**");
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ subscriptionInfo.createTopic(new CreateTopicPlan(initialTopicMeta)).getCode());
+
+ final Map updatedAttributes = new HashMap<>();
+ updatedAttributes.put(TopicConstant.OWNER_ID_KEY, "owner2");
+ updatedAttributes.put(TopicConstant.OWNER_EPOCH_KEY, "6");
+ final TSStatus alterStatus =
+ subscriptionInfo.alterTopic(
+ new AlterTopicPlan(initialTopicMeta.deepCopyWithUpdatedAttributes(updatedAttributes)));
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), alterStatus.getCode());
+ Assert.assertEquals("owner2", subscriptionInfo.getTopicMeta(topicName).getOwnerId());
+ Assert.assertEquals(6L, subscriptionInfo.getTopicMeta(topicName).getOwnerEpoch());
+ Assert.assertEquals(
+ "root.sg.**",
+ subscriptionInfo.getTopicMeta(topicName).getConfig().getString(TopicConstant.PATH_KEY));
+ }
+
+ @Test
+ public void testCollectTopicOwnerLeaseEntriesSkipsBlockedTopic() {
+ final String topicName = "topic-" + UUID.randomUUID();
+ final String blockedTopicName = "topic-" + UUID.randomUUID();
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+
+ final long ownerLeaseDurationMs = 60000;
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ subscriptionInfo
+ .createTopic(
+ new CreateTopicPlan(createTopicMeta(topicName, "owner1", 5L, ownerLeaseDurationMs)))
+ .getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ subscriptionInfo
+ .createTopic(
+ new CreateTopicPlan(
+ createTopicMeta(blockedTopicName, "owner1", 5L, ownerLeaseDurationMs)))
+ .getCode());
+
+ final List entries =
+ subscriptionInfo.collectTopicOwnerLeaseEntries(Collections.singleton(blockedTopicName));
+
+ Assert.assertEquals(1, entries.size());
+ final TTopicOwnerLeaseEntry entry = entries.get(0);
+ Assert.assertEquals(topicName, entry.getTopicName());
+ Assert.assertEquals("owner1", entry.getOwnerId());
+ Assert.assertEquals(5L, entry.getOwnerEpoch());
+ // Relative remaining duration equals the configured lease duration; read-only, no mutation.
+ Assert.assertEquals(ownerLeaseDurationMs, entry.getLeaseRemainingMs());
+ }
+
+ @Test
+ public void testAlterTopicOwnerAndShowTopicOwner() {
+ final String topicName = "topic-" + UUID.randomUUID();
+ final long ownerLeaseDurationMs = 123456L;
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+
+ final TopicMeta initialTopicMeta = createTopicMeta(topicName, "owner1", 5L);
+ initialTopicMeta.getConfig().getAttribute().put(TopicConstant.PATH_KEY, "root.sg.**");
+ initialTopicMeta.getConfig().getAttribute().put(TopicConstant.START_TIME_KEY, "0");
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ subscriptionInfo.createTopic(new CreateTopicPlan(initialTopicMeta)).getCode());
+
+ final Map updatedAttributes = new HashMap<>();
+ updatedAttributes.put(TopicConstant.OWNER_ID_KEY, "owner2");
+ updatedAttributes.put(TopicConstant.OWNER_EPOCH_KEY, "6");
+ updatedAttributes.put(
+ TopicConstant.OWNER_LEASE_DURATION_MS_KEY, String.valueOf(ownerLeaseDurationMs));
+ final TopicMeta updatedTopicMeta =
+ initialTopicMeta.deepCopyWithUpdatedAttributes(updatedAttributes);
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ subscriptionInfo.alterTopic(new AlterTopicPlan(updatedTopicMeta)).getCode());
+
+ final TShowTopicResp showTopicResp =
+ ((TopicTableResp) subscriptionInfo.showTopics()).convertToTShowTopicResp();
+
+ Assert.assertEquals(TSStatusCode.SUCCESS_STATUS.getStatusCode(), showTopicResp.status.code);
+ Assert.assertEquals(1, showTopicResp.getTopicInfoListSize());
+
+ final TShowTopicInfo showTopicInfo = showTopicResp.getTopicInfoList().get(0);
+ Assert.assertEquals(topicName, showTopicInfo.getTopicName());
+ Assert.assertEquals(1L, showTopicInfo.getCreationTime());
+ Assert.assertTrue(showTopicInfo.getTopicAttributes().contains("path=root.sg.**"));
+ Assert.assertTrue(showTopicInfo.getTopicAttributes().contains("start-time=0"));
+ Assert.assertTrue(showTopicInfo.getTopicAttributes().contains("owner-id=owner2"));
+ Assert.assertTrue(showTopicInfo.getTopicAttributes().contains("owner-epoch=6"));
+ Assert.assertTrue(
+ showTopicInfo
+ .getTopicAttributes()
+ .contains("owner-lease-duration-ms=" + ownerLeaseDurationMs));
+ // The absolute lease expiry is DataNode-local runtime state; it is never a topic attribute and
+ // must not appear in SHOW TOPICS output.
+ Assert.assertFalse(showTopicInfo.getTopicAttributes().contains("owner-lease-expire-time-ms"));
+ }
+
+ private TopicMeta createTopicMeta(
+ final String topicName, final String ownerId, final long ownerEpoch) {
+ final Map topicAttributes = new HashMap<>();
+ topicAttributes.put(TopicConstant.OWNER_ID_KEY, ownerId);
+ topicAttributes.put(TopicConstant.OWNER_EPOCH_KEY, String.valueOf(ownerEpoch));
+ return new TopicMeta(topicName, 1, topicAttributes);
+ }
+
+ private TopicMeta createTopicMeta(
+ final String topicName,
+ final String ownerId,
+ final long ownerEpoch,
+ final long ownerLeaseDurationMs) {
+ final Map topicAttributes = new HashMap<>();
+ topicAttributes.put(TopicConstant.OWNER_ID_KEY, ownerId);
+ topicAttributes.put(TopicConstant.OWNER_EPOCH_KEY, String.valueOf(ownerEpoch));
+ topicAttributes.put(
+ TopicConstant.OWNER_LEASE_DURATION_MS_KEY, String.valueOf(ownerLeaseDurationMs));
+ return new TopicMeta(topicName, 1, topicAttributes);
+ }
+}
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 144b1e9530da1..61eaa8a4c4c64 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
@@ -20,6 +20,7 @@
package org.apache.iotdb.confignode.persistence.subscription;
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.confignode.consensus.request.write.subscription.topic.CreateTopicPlan;
import org.apache.iotdb.confignode.rpc.thrift.TCreateTopicReq;
@@ -189,6 +190,33 @@ public void testRejectConsensusOnlyRetentionOnLiveTopic() {
assertCreateRejected(subscriptionInfo, attributes, "only supported for consensus topics");
}
+ @Test
+ public void testRejectOwnerLeaseDurationBelowMin() {
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map attributes = new HashMap<>();
+ attributes.put(TopicConstant.OWNER_ID_KEY, "owner1");
+ attributes.put(TopicConstant.OWNER_EPOCH_KEY, "1");
+ // Well below the default 1-minute floor.
+ attributes.put(TopicConstant.OWNER_LEASE_DURATION_MS_KEY, "5000");
+
+ assertCreateRejected(subscriptionInfo, attributes, "below the minimum allowed");
+ }
+
+ @Test
+ public void testAcceptOwnerLeaseDurationAtMin() throws Exception {
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map attributes = new HashMap<>();
+ attributes.put(TopicConstant.OWNER_ID_KEY, "owner1");
+ attributes.put(TopicConstant.OWNER_EPOCH_KEY, "1");
+ attributes.put(
+ TopicConstant.OWNER_LEASE_DURATION_MS_KEY,
+ String.valueOf(SubscriptionConfig.getInstance().getSubscriptionOwnerLeaseDurationMsMin()));
+
+ Assert.assertTrue(
+ subscriptionInfo.validateBeforeCreatingTopic(
+ new TCreateTopicReq("owner_topic").setTopicAttributes(attributes)));
+ }
+
private static Map newConsensusTableTopicAttributes() {
final Map attributes = new HashMap<>();
attributes.put(SystemConstant.SQL_DIALECT_KEY, SystemConstant.SQL_DIALECT_TABLE_VALUE);
diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
index e4a780d13f890..8eb26f05e0a69 100644
--- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
+++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
@@ -184,6 +184,8 @@ private DataNodeMiscMessages() {}
"Error occurred when pushing single topic meta";
public static final String ERROR_PUSHING_MULTI_TOPIC_META =
"Error occurred when pushing multi topic meta";
+ public static final String ERROR_PUSHING_TOPIC_OWNER_LEASE =
+ "Error occurred when pushing topic owner lease";
public static final String ERROR_PUSHING_CONSUMER_GROUP_META =
"Error occurred when pushing consumer group meta";
public static final String ERROR_PUSHING_SINGLE_CONSUMER_GROUP_META =
diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
index 964163ce14fbc..265fc223375e4 100644
--- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
+++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
@@ -551,6 +551,8 @@ public final class DataNodeQueryMessages {
"Not support for this sql in START PIPE, please enter pipename.";
public static final String NOT_SUPPORT_FOR_THIS_SQL_IN_STOP_PIPE =
"Not support for this sql in STOP PIPE, please enter pipename.";
+ public static final String NOT_SUPPORT_FOR_THIS_SQL_IN_ALTER_TOPIC =
+ "Not support for this sql in ALTER TOPIC, please enter topicName.";
public static final String GET_REGION_ID_STATEMENT_EXPRESSION_MUST_BE_A =
"Get region id statement‘ expression must be a time expression";
public static final String WRONG_SPACE_QUOTA_TYPE =
diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
index 14e9b3f2c0000..69e4370b6645b 100644
--- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
+++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeMiscMessages.java
@@ -183,6 +183,8 @@ private DataNodeMiscMessages() {}
"推送单个 Topic 元数据时发生错误";
public static final String ERROR_PUSHING_MULTI_TOPIC_META =
"推送多个 Topic 元数据时发生错误";
+ public static final String ERROR_PUSHING_TOPIC_OWNER_LEASE =
+ "推送 Topic owner 租约时发生错误";
public static final String ERROR_PUSHING_CONSUMER_GROUP_META =
"推送消费者组元数据时发生错误";
public static final String ERROR_PUSHING_SINGLE_CONSUMER_GROUP_META =
diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
index ae020f8c5a31f..559eaea448027 100644
--- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
+++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodeQueryMessages.java
@@ -550,6 +550,8 @@ public final class DataNodeQueryMessages {
"START PIPE 不支持此 SQL,请输入管道名。";
public static final String NOT_SUPPORT_FOR_THIS_SQL_IN_STOP_PIPE =
"STOP PIPE 不支持此 SQL,请输入管道名。";
+ public static final String NOT_SUPPORT_FOR_THIS_SQL_IN_ALTER_TOPIC =
+ "ALTER TOPIC 不支持此 SQL,请输入 topic 名。";
public static final String GET_REGION_ID_STATEMENT_EXPRESSION_MUST_BE_A =
"GET REGION ID 语句的表达式必须是时间表达式";
public static final String WRONG_SPACE_QUOTA_TYPE =
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
index b21d3a3a79209..e7e50d8b9e502 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/client/ConfigNodeClient.java
@@ -59,6 +59,7 @@
import org.apache.iotdb.confignode.rpc.thrift.TAlterPipeReq;
import org.apache.iotdb.confignode.rpc.thrift.TAlterSchemaTemplateReq;
import org.apache.iotdb.confignode.rpc.thrift.TAlterTimeSeriesReq;
+import org.apache.iotdb.confignode.rpc.thrift.TAlterTopicReq;
import org.apache.iotdb.confignode.rpc.thrift.TAuthizedPatternTreeResp;
import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerRelationalReq;
import org.apache.iotdb.confignode.rpc.thrift.TAuthorizerReq;
@@ -1209,6 +1210,12 @@ public TSStatus createTopic(TCreateTopicReq req) throws TException {
() -> client.createTopic(req), status -> !updateConfigNodeLeader(status));
}
+ @Override
+ public TSStatus alterTopic(TAlterTopicReq req) throws TException {
+ return executeRemoteCallWithRetry(
+ () -> client.alterTopic(req), status -> !updateConfigNodeLeader(status));
+ }
+
@Override
public TSStatus dropTopic(String topicName) throws TException {
return executeRemoteCallWithRetry(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
index 7890802f36243..fc8a0b37bcaf6 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
@@ -308,6 +308,7 @@
import org.apache.iotdb.mpp.rpc.thrift.TPushTopicMetaReq;
import org.apache.iotdb.mpp.rpc.thrift.TPushTopicMetaResp;
import org.apache.iotdb.mpp.rpc.thrift.TPushTopicMetaRespExceptionMessage;
+import org.apache.iotdb.mpp.rpc.thrift.TPushTopicOwnerLeaseReq;
import org.apache.iotdb.mpp.rpc.thrift.TRegionLeaderChangeReq;
import org.apache.iotdb.mpp.rpc.thrift.TRegionLeaderChangeResp;
import org.apache.iotdb.mpp.rpc.thrift.TRegionMigrateResult;
@@ -1546,6 +1547,20 @@ public TPushTopicMetaResp pushMultiTopicMeta(TPushMultiTopicMetaReq req) {
}
}
+ @Override
+ public TSStatus pushTopicOwnerLease(TPushTopicOwnerLeaseReq req) {
+ if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ }
+ try {
+ SubscriptionAgent.topic().handleTopicOwnerLeases(req.getOwnerLeases());
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ } catch (Exception e) {
+ LOGGER.warn(DataNodeMiscMessages.ERROR_PUSHING_TOPIC_OWNER_LEASE, e);
+ return new TSStatus(TSStatusCode.TOPIC_PUSH_META_ERROR.getStatusCode());
+ }
+ }
+
@Override
public TPushConsumerGroupMetaResp pushConsumerGroupMeta(TPushConsumerGroupMetaReq req) {
if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java
index 827590e86a011..4eff93bfc12d2 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TableConfigTaskVisitor.java
@@ -140,6 +140,7 @@
import org.apache.iotdb.db.queryengine.plan.execution.config.sys.pipe.ShowPipeTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.sys.pipe.StartPipeTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.sys.pipe.StopPipeTask;
+import org.apache.iotdb.db.queryengine.plan.execution.config.sys.subscription.AlterTopicTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.sys.subscription.CreateTopicTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.sys.subscription.DropSubscriptionTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.sys.subscription.DropTopicTask;
@@ -154,6 +155,7 @@
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AlterColumnDataType;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AlterDB;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AlterPipe;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AlterTopic;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AstVisitor;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ClearCache;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ColumnDefinition;
@@ -1392,6 +1394,17 @@ public IConfigTask visitCreateTopic(CreateTopic node, MPPQueryContext context) {
return new CreateTopicTask(node);
}
+ @Override
+ public IConfigTask visitAlterTopic(AlterTopic node, MPPQueryContext context) {
+ context.setQueryType(QueryType.OTHER);
+ accessControl.checkUserGlobalSysPrivilege(context);
+
+ node.getTopicAttributes()
+ .put(SystemConstant.SQL_DIALECT_KEY, SystemConstant.SQL_DIALECT_TABLE_VALUE);
+
+ return new AlterTopicTask(node);
+ }
+
@Override
public IConfigTask visitDropTopic(DropTopic node, MPPQueryContext context) {
context.setQueryType(QueryType.OTHER);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TreeConfigTaskVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TreeConfigTaskVisitor.java
index 8b7792bca985a..28745712d220c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TreeConfigTaskVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/TreeConfigTaskVisitor.java
@@ -123,6 +123,7 @@
import org.apache.iotdb.db.queryengine.plan.execution.config.sys.quota.SetThrottleQuotaTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.sys.quota.ShowSpaceQuotaTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.sys.quota.ShowThrottleQuotaTask;
+import org.apache.iotdb.db.queryengine.plan.execution.config.sys.subscription.AlterTopicTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.sys.subscription.CreateTopicTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.sys.subscription.DropSubscriptionTask;
import org.apache.iotdb.db.queryengine.plan.execution.config.sys.subscription.DropTopicTask;
@@ -192,6 +193,7 @@
import org.apache.iotdb.db.queryengine.plan.statement.metadata.region.MigrateRegionStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.region.ReconstructRegionStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.region.RemoveRegionStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.AlterTopicStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.CreateTopicStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.DropSubscriptionStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.DropTopicStatement;
@@ -730,6 +732,16 @@ public IConfigTask visitCreateTopic(
return new CreateTopicTask(createTopicStatement);
}
+ @Override
+ public IConfigTask visitAlterTopic(
+ AlterTopicStatement alterTopicStatement, MPPQueryContext context) {
+ alterTopicStatement
+ .getTopicAttributes()
+ .put(SystemConstant.SQL_DIALECT_KEY, SystemConstant.SQL_DIALECT_TREE_VALUE);
+
+ return new AlterTopicTask(alterTopicStatement);
+ }
+
@Override
public IConfigTask visitDropTopic(
DropTopicStatement dropTopicStatement, MPPQueryContext context) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
index 5b8a79cda9a1e..6eda402c2eb2c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java
@@ -106,6 +106,7 @@
import org.apache.iotdb.confignode.rpc.thrift.TAlterPipeReq;
import org.apache.iotdb.confignode.rpc.thrift.TAlterSchemaTemplateReq;
import org.apache.iotdb.confignode.rpc.thrift.TAlterTimeSeriesReq;
+import org.apache.iotdb.confignode.rpc.thrift.TAlterTopicReq;
import org.apache.iotdb.confignode.rpc.thrift.TCountDatabaseResp;
import org.apache.iotdb.confignode.rpc.thrift.TCountTimeSlotListReq;
import org.apache.iotdb.confignode.rpc.thrift.TCountTimeSlotListResp;
@@ -291,6 +292,7 @@
import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.ShowPipesStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.StartPipeStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.StopPipeStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.AlterTopicStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.CreateTopicStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.DropSubscriptionStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.DropTopicStatement;
@@ -2882,6 +2884,34 @@ public SettableFuture createTopic(
return future;
}
+ @Override
+ public SettableFuture alterTopic(
+ final AlterTopicStatement alterTopicStatement) {
+ if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
+ return SUBSCRIPTION_NOT_ENABLED_ERROR_FUTURE;
+ }
+
+ final SettableFuture future = SettableFuture.create();
+
+ try (final ConfigNodeClient configNodeClient =
+ CONFIG_NODE_CLIENT_MANAGER.borrowClient(ConfigNodeInfo.CONFIG_REGION_ID)) {
+ final TSStatus tsStatus =
+ configNodeClient.alterTopic(
+ new TAlterTopicReq()
+ .setTopicName(alterTopicStatement.getTopicName())
+ .setTopicAttributes(alterTopicStatement.getTopicAttributes())
+ .setSubscribedConsumerGroupIds(Collections.emptySet()));
+ if (TSStatusCode.SUCCESS_STATUS.getStatusCode() != tsStatus.getCode()) {
+ future.setException(new IoTDBException(tsStatus));
+ } else {
+ future.set(new ConfigTaskResult(TSStatusCode.SUCCESS_STATUS));
+ }
+ } catch (Exception e) {
+ future.setException(e);
+ }
+ return future;
+ }
+
@Override
public SettableFuture dropTopic(final DropTopicStatement dropTopicStatement) {
if (!SubscriptionConfig.getInstance().getSubscriptionEnabled()) {
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/IConfigTaskExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/IConfigTaskExecutor.java
index b45209aeffed8..4e551983d735d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/IConfigTaskExecutor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/IConfigTaskExecutor.java
@@ -76,6 +76,7 @@
import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.ShowPipesStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.StartPipeStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.pipe.StopPipeStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.AlterTopicStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.CreateTopicStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.DropSubscriptionStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.DropTopicStatement;
@@ -243,6 +244,8 @@ SettableFuture dropSubscription(
SettableFuture createTopic(CreateTopicStatement createTopicStatement);
+ SettableFuture alterTopic(AlterTopicStatement alterTopicStatement);
+
SettableFuture dropTopic(DropTopicStatement dropTopicStatement);
SettableFuture showTopics(ShowTopicsStatement showTopicsStatement);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/subscription/AlterTopicTask.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/subscription/AlterTopicTask.java
new file mode 100644
index 0000000000000..33ae342267be2
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/sys/subscription/AlterTopicTask.java
@@ -0,0 +1,49 @@
+/*
+ * 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.queryengine.plan.execution.config.sys.subscription;
+
+import org.apache.iotdb.db.queryengine.plan.execution.config.ConfigTaskResult;
+import org.apache.iotdb.db.queryengine.plan.execution.config.IConfigTask;
+import org.apache.iotdb.db.queryengine.plan.execution.config.executor.IConfigTaskExecutor;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AlterTopic;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.AlterTopicStatement;
+
+import com.google.common.util.concurrent.ListenableFuture;
+
+public class AlterTopicTask implements IConfigTask {
+
+ private final AlterTopicStatement alterTopicStatement;
+
+ public AlterTopicTask(final AlterTopicStatement alterTopicStatement) {
+ this.alterTopicStatement = alterTopicStatement;
+ }
+
+ public AlterTopicTask(final AlterTopic alterTopic) {
+ this.alterTopicStatement = new AlterTopicStatement();
+ this.alterTopicStatement.setTopicName(alterTopic.getTopicName());
+ this.alterTopicStatement.setTopicAttributes(alterTopic.getTopicAttributes());
+ }
+
+ @Override
+ public ListenableFuture execute(final IConfigTaskExecutor configTaskExecutor)
+ throws InterruptedException {
+ return configTaskExecutor.alterTopic(alterTopicStatement);
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/ASTVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/ASTVisitor.java
index 49f79a6a923c0..6ecf65c86bb3c 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/ASTVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/parser/ASTVisitor.java
@@ -214,6 +214,7 @@
import org.apache.iotdb.db.queryengine.plan.statement.metadata.region.MigrateRegionStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.region.ReconstructRegionStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.region.RemoveRegionStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.AlterTopicStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.CreateTopicStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.DropSubscriptionStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.DropTopicStatement;
@@ -4436,6 +4437,22 @@ public Statement visitCreateTopic(IoTDBSqlParser.CreateTopicContext ctx) {
return createTopicStatement;
}
+ @Override
+ public Statement visitAlterTopic(IoTDBSqlParser.AlterTopicContext ctx) {
+ final AlterTopicStatement alterTopicStatement = new AlterTopicStatement();
+
+ if (ctx.topicName != null) {
+ alterTopicStatement.setTopicName(parseIdentifier(ctx.topicName.getText()));
+ } else {
+ throw new SemanticException(DataNodeQueryMessages.NOT_SUPPORT_FOR_THIS_SQL_IN_ALTER_TOPIC);
+ }
+
+ alterTopicStatement.setTopicAttributes(
+ parseTopicAttributesClause(ctx.topicAttributesClause().topicAttributeClause()));
+
+ return alterTopicStatement;
+ }
+
private Map parseTopicAttributesClause(
List contexts) {
final Map collectorMap = new HashMap<>();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java
index 8912ac37264a4..81c3c24a7ccb5 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/analyzer/StatementAnalyzer.java
@@ -147,6 +147,7 @@
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AddColumn;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AlterDB;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AlterPipe;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AlterTopic;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AsofJoinOn;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AstVisitor;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CopyTo;
@@ -4859,6 +4860,11 @@ public Scope visitCreateTopic(CreateTopic node, Optional context) {
return createAndAssignScope(node, context);
}
+ @Override
+ public Scope visitAlterTopic(AlterTopic node, Optional context) {
+ return createAndAssignScope(node, context);
+ }
+
@Override
public Scope visitDropTopic(DropTopic node, Optional context) {
return createAndAssignScope(node, context);
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/security/TreeAccessCheckVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/security/TreeAccessCheckVisitor.java
index fd9757c5f8902..41f91bb461809 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/security/TreeAccessCheckVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/security/TreeAccessCheckVisitor.java
@@ -122,6 +122,7 @@
import org.apache.iotdb.db.queryengine.plan.statement.metadata.region.MigrateRegionStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.region.ReconstructRegionStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.region.RemoveRegionStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.AlterTopicStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.CreateTopicStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.DropSubscriptionStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.DropTopicStatement;
@@ -884,6 +885,12 @@ public TSStatus visitCreateTopic(CreateTopicStatement statement, TreeAccessCheck
context.setAuditLogOperation(AuditLogOperation.DDL), statement::getTopicName);
}
+ @Override
+ public TSStatus visitAlterTopic(AlterTopicStatement statement, TreeAccessCheckContext context) {
+ return checkPipeManagement(
+ context.setAuditLogOperation(AuditLogOperation.DDL), statement::getTopicName);
+ }
+
@Override
public TSStatus visitShowTopics(ShowTopicsStatement statement, TreeAccessCheckContext context) {
return checkPipeManagement(
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AlterTopic.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AlterTopic.java
new file mode 100644
index 0000000000000..8598c1525266d
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AlterTopic.java
@@ -0,0 +1,92 @@
+/*
+ * 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.queryengine.plan.relational.sql.ast;
+
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.AstMemoryEstimationHelper;
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.IAstVisitor;
+
+import org.apache.tsfile.utils.RamUsageEstimator;
+
+import java.util.Map;
+import java.util.Objects;
+
+import static com.google.common.base.MoreObjects.toStringHelper;
+import static java.util.Objects.requireNonNull;
+
+public class AlterTopic extends SubscriptionStatement {
+ private static final long INSTANCE_SIZE =
+ RamUsageEstimator.shallowSizeOfInstance(AlterTopic.class);
+
+ private final String topicName;
+ private final Map topicAttributes;
+
+ public AlterTopic(final String topicName, final Map topicAttributes) {
+ this.topicName = requireNonNull(topicName, "topic name can not be null");
+ this.topicAttributes = requireNonNull(topicAttributes, "topic attributes can not be null");
+ }
+
+ public String getTopicName() {
+ return topicName;
+ }
+
+ public Map getTopicAttributes() {
+ return topicAttributes;
+ }
+
+ @Override
+ public R accept(final IAstVisitor visitor, final C context) {
+ return ((AstVisitor) visitor).visitAlterTopic(this, context);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(topicName, topicAttributes);
+ }
+
+ @Override
+ public boolean equals(final Object obj) {
+ if (this == obj) {
+ return true;
+ }
+ if (obj == null || getClass() != obj.getClass()) {
+ return false;
+ }
+ final AlterTopic that = (AlterTopic) obj;
+ return Objects.equals(this.topicName, that.topicName)
+ && Objects.equals(this.topicAttributes, that.topicAttributes);
+ }
+
+ @Override
+ public String toString() {
+ return toStringHelper(this)
+ .add("topicName", topicName)
+ .add("topicAttributes", topicAttributes)
+ .toString();
+ }
+
+ @Override
+ public long ramBytesUsed() {
+ long size = INSTANCE_SIZE;
+ size += AstMemoryEstimationHelper.getEstimatedSizeOfNodeLocation(getLocationInternal());
+ size += RamUsageEstimator.sizeOf(topicName);
+ size += RamUsageEstimator.sizeOfMap(topicAttributes);
+ return size;
+ }
+}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AstVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AstVisitor.java
index 7c311fa7f62b6..c00b7f4d492a8 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AstVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/ast/AstVisitor.java
@@ -343,6 +343,10 @@ default R visitCreateTopic(CreateTopic node, C context) {
return visitStatement(node, context);
}
+ default R visitAlterTopic(AlterTopic node, C context) {
+ return visitStatement(node, context);
+ }
+
default R visitDropTopic(DropTopic node, C context) {
return visitStatement(node, context);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
index a19168b50f42f..43dc3ccc31b64 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/AstBuilder.java
@@ -159,6 +159,7 @@
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AlterColumnDataType;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AlterDB;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AlterPipe;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AlterTopic;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AsofJoinOn;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ClearCache;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ColumnDefinition;
@@ -1391,6 +1392,13 @@ public Node visitCreateTopicStatement(RelationalSqlParser.CreateTopicStatementCo
return new CreateTopic(topicName, hasIfNotExistsCondition, topicAttributes);
}
+ @Override
+ public Node visitAlterTopicStatement(RelationalSqlParser.AlterTopicStatementContext ctx) {
+ final String topicName = ((Identifier) visit(ctx.identifier())).getValue();
+ return new AlterTopic(
+ topicName, parseTopicAttributesClause(ctx.topicAttributesClause().topicAttributeClause()));
+ }
+
private Map parseTopicAttributesClause(
List contexts) {
final Map tppicMap = new HashMap<>();
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/util/DataNodeSqlFormatter.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/util/DataNodeSqlFormatter.java
index b806fd725928e..887a6082a5d5d 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/util/DataNodeSqlFormatter.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/relational/sql/util/DataNodeSqlFormatter.java
@@ -29,6 +29,7 @@
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AddColumn;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AlterDB;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AlterPipe;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AlterTopic;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AstVisitor;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.ColumnDefinition;
import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.CopyTo;
@@ -771,6 +772,31 @@ public Void visitCreateTopic(CreateTopic node, Integer context) {
return null;
}
+ @Override
+ public Void visitAlterTopic(AlterTopic node, Integer context) {
+ builder.append("ALTER TOPIC ");
+ builder.append(node.getTopicName());
+ builder.append(" \n");
+
+ builder
+ .append("WITH (")
+ .append("\n")
+ .append(
+ node.getTopicAttributes().entrySet().stream()
+ .map(
+ entry ->
+ indentString(1)
+ + "\""
+ + entry.getKey()
+ + "\" = \""
+ + entry.getValue()
+ + "\"")
+ .collect(joining(", " + "\n")))
+ .append(")\n");
+
+ return null;
+ }
+
@Override
public Void visitDropTopic(DropTopic node, Integer context) {
builder.append("DROP TOPIC ");
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementType.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementType.java
index b40c6444816fc..88a2959f1475b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementType.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementType.java
@@ -184,6 +184,7 @@ public enum StatementType {
REPAIR_DATA_PARTITION_TABLE,
CREATE_TOPIC,
+ ALTER_TOPIC,
DROP_TOPIC,
SHOW_TOPICS,
SHOW_SUBSCRIPTIONS,
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementVisitor.java
index 847e850c52172..39af259928b6b 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementVisitor.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/StatementVisitor.java
@@ -107,6 +107,7 @@
import org.apache.iotdb.db.queryengine.plan.statement.metadata.region.MigrateRegionStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.region.ReconstructRegionStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.region.RemoveRegionStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.AlterTopicStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.CreateTopicStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.DropSubscriptionStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.DropTopicStatement;
@@ -644,6 +645,10 @@ public R visitCreateTopic(CreateTopicStatement createTopicStatement, C context)
return visitStatement(createTopicStatement, context);
}
+ public R visitAlterTopic(AlterTopicStatement alterTopicStatement, C context) {
+ return visitStatement(alterTopicStatement, context);
+ }
+
public R visitDropTopic(DropTopicStatement dropTopicStatement, C context) {
return visitStatement(dropTopicStatement, context);
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/subscription/AlterTopicStatement.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/subscription/AlterTopicStatement.java
new file mode 100644
index 0000000000000..dd06b3034b00b
--- /dev/null
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/statement/metadata/subscription/AlterTopicStatement.java
@@ -0,0 +1,73 @@
+/*
+ * 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.queryengine.plan.statement.metadata.subscription;
+
+import org.apache.iotdb.commons.path.PartialPath;
+import org.apache.iotdb.db.queryengine.plan.analyze.QueryType;
+import org.apache.iotdb.db.queryengine.plan.statement.IConfigStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.Statement;
+import org.apache.iotdb.db.queryengine.plan.statement.StatementType;
+import org.apache.iotdb.db.queryengine.plan.statement.StatementVisitor;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+public class AlterTopicStatement extends Statement implements IConfigStatement {
+
+ private String topicName;
+ private Map topicAttributes;
+
+ public AlterTopicStatement() {
+ super();
+ statementType = StatementType.ALTER_TOPIC;
+ }
+
+ public String getTopicName() {
+ return topicName;
+ }
+
+ public Map getTopicAttributes() {
+ return topicAttributes;
+ }
+
+ public void setTopicName(final String topicName) {
+ this.topicName = topicName;
+ }
+
+ public void setTopicAttributes(final Map topicAttributes) {
+ this.topicAttributes = topicAttributes;
+ }
+
+ @Override
+ public QueryType getQueryType() {
+ return QueryType.OTHER;
+ }
+
+ @Override
+ public List getPaths() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public R accept(final StatementVisitor visitor, final C context) {
+ return visitor.visitAlterTopic(this, context);
+ }
+}
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 e89578fe556e6..485e553e8ee0a 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
@@ -19,10 +19,15 @@
package org.apache.iotdb.db.subscription.agent;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta;
import org.apache.iotdb.commons.subscription.meta.topic.TopicMetaKeeper;
import org.apache.iotdb.db.i18n.DataNodeMiscMessages;
import org.apache.iotdb.mpp.rpc.thrift.TPushTopicMetaRespExceptionMessage;
+import org.apache.iotdb.mpp.rpc.thrift.TTopicOwnerLeaseEntry;
+import org.apache.iotdb.rpc.RpcUtils;
+import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.rpc.subscription.config.ConsumerConfig;
import org.apache.iotdb.rpc.subscription.config.TopicConfig;
import org.apache.iotdb.rpc.subscription.config.TopicConstant;
@@ -31,6 +36,7 @@
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Set;
import java.util.stream.Collectors;
@@ -87,6 +93,8 @@ public TPushTopicMetaRespExceptionMessage handleSingleTopicMetaChanges(
private void handleSingleTopicMetaChangesInternal(final TopicMeta metaFromCoordinator) {
final String topicName = metaFromCoordinator.getTopicName();
+ TopicMeta.validateOwnerProgression(
+ topicMetaKeeper.getTopicMeta(topicName), metaFromCoordinator);
topicMetaKeeper.removeTopicMeta(topicName);
topicMetaKeeper.addTopicMeta(topicName, metaFromCoordinator);
SubscriptionAgent.broker()
@@ -196,4 +204,94 @@ public Map getTopicConfigs(final Set topicNames) {
releaseReadLock();
}
}
+
+ public TSStatus checkTopicOwner(final ConsumerConfig consumerConfig, final String topicName) {
+ acquireReadLock();
+ try {
+ final TopicMeta topicMeta = topicMetaKeeper.getTopicMeta(topicName);
+ if (Objects.isNull(topicMeta) || !topicMeta.isOwnerFencingEnabled()) {
+ return RpcUtils.SUCCESS_STATUS;
+ }
+
+ final String requestOwnerId = consumerConfig.getOwnerId();
+ if (Objects.isNull(requestOwnerId)) {
+ return RpcUtils.getStatus(
+ TSStatusCode.SUBSCRIPTION_OWNER_REQUIRED,
+ String.format(
+ "Subscription: topic %s enables owner fencing, but consumer %s does not carry owner-id.",
+ topicName, consumerConfig));
+ }
+
+ final Long requestOwnerEpoch = consumerConfig.getOwnerEpoch();
+ if (Objects.isNull(requestOwnerEpoch)) {
+ return RpcUtils.getStatus(
+ TSStatusCode.SUBSCRIPTION_OWNER_EPOCH_REQUIRED,
+ String.format(
+ "Subscription: topic %s enables owner fencing, but consumer %s does not carry owner-epoch.",
+ topicName, consumerConfig));
+ }
+
+ if (Objects.nonNull(topicMeta.getOwnerLeaseExpireTimeMs())
+ && System.currentTimeMillis() > topicMeta.getOwnerLeaseExpireTimeMs()) {
+ return RpcUtils.getStatus(
+ TSStatusCode.SUBSCRIPTION_OWNER_LEASE_EXPIRED,
+ String.format(
+ "Subscription: owner lease for topic %s has expired, owner-id: %s, owner-epoch: %s.",
+ topicName, topicMeta.getOwnerId(), topicMeta.getOwnerEpoch()));
+ }
+
+ if (!topicMeta.matchesOwner(requestOwnerId, requestOwnerEpoch)) {
+ return RpcUtils.getStatus(
+ TSStatusCode.SUBSCRIPTION_OWNER_FENCED,
+ String.format(
+ "Subscription: consumer owner is fenced for topic %s, request owner-id: %s,"
+ + " request owner-epoch: %s, current owner-id: %s, current owner-epoch: %s.",
+ topicName,
+ requestOwnerId,
+ requestOwnerEpoch,
+ topicMeta.getOwnerId(),
+ topicMeta.getOwnerEpoch()));
+ }
+
+ return RpcUtils.SUCCESS_STATUS;
+ } finally {
+ releaseReadLock();
+ }
+ }
+
+ public TSStatus checkTopicOwners(
+ final ConsumerConfig consumerConfig, final Iterable topicNames) {
+ for (final String topicName : topicNames) {
+ final TSStatus status = checkTopicOwner(consumerConfig, topicName);
+ if (status.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return status;
+ }
+ }
+ return RpcUtils.SUCCESS_STATUS;
+ }
+
+ /**
+ * Apply owner lease renewals pushed by ConfigNode via the dedicated subscription owner heartbeat.
+ * The pushed remaining duration is converted to a DataNode-local expire time on the local clock,
+ * so no absolute timestamp is compared across nodes. Owner identity/epoch changes are delivered
+ * via the topic-meta push path; here we only refresh the lease for the matching current owner.
+ */
+ public void handleTopicOwnerLeases(final List ownerLeases) {
+ if (Objects.isNull(ownerLeases) || ownerLeases.isEmpty()) {
+ return;
+ }
+ acquireWriteLock();
+ try {
+ for (final TTopicOwnerLeaseEntry lease : ownerLeases) {
+ final TopicMeta topicMeta = topicMetaKeeper.getTopicMeta(lease.getTopicName());
+ if (Objects.isNull(topicMeta)) {
+ continue;
+ }
+ topicMeta.applyOwnerLeaseFromHeartbeat(
+ lease.getOwnerId(), lease.getOwnerEpoch(), lease.getLeaseRemainingMs());
+ }
+ } finally {
+ releaseWriteLock();
+ }
+ }
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java
index 3c76d09c88dd5..134d6f7b90db5 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1.java
@@ -347,15 +347,22 @@ private TPipeSubscribeResp handleSubscriptionHeartbeatInternal(final Subscriptio
processorBufferedCommitContexts.size());
}
+ final Set subscribedTopicNames =
+ SubscriptionAgent.consumer()
+ .getTopicNamesSubscribedByConsumer(
+ consumerConfig.getConsumerGroupId(), consumerConfig.getConsumerId());
+
+ final TSStatus ownerStatus =
+ SubscriptionAgent.topic().checkTopicOwners(consumerConfig, subscribedTopicNames);
+ if (ownerStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return PipeSubscribeHeartbeatResp.toTPipeSubscribeResp(ownerStatus);
+ }
+
LOGGER.info(DataNodeMiscMessages.SUBSCRIPTION_CONSUMER_HEARTBEAT_SUCCESS, consumerConfig);
// fetch subscribed topics
final Map topics =
- SubscriptionAgent.topic()
- .getTopicConfigs(
- SubscriptionAgent.consumer()
- .getTopicNamesSubscribedByConsumer(
- consumerConfig.getConsumerGroupId(), consumerConfig.getConsumerId()));
+ SubscriptionAgent.topic().getTopicConfigs(subscribedTopicNames);
// fetch available endpoints
final Map endPoints = new HashMap<>();
@@ -426,6 +433,11 @@ private TPipeSubscribeResp handlePipeSubscribeSubscribeInternal(
// subscribe topics
final Set topicNames = req.getTopicNames();
+ final TSStatus ownerStatus =
+ SubscriptionAgent.topic().checkTopicOwners(consumerConfig, topicNames);
+ if (ownerStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return PipeSubscribeSubscribeResp.toTPipeSubscribeResp(ownerStatus);
+ }
subscribe(consumerConfig, topicNames);
LOGGER.info(
@@ -518,6 +530,18 @@ private TPipeSubscribeResp handlePipeSubscribePollInternal(final PipeSubscribePo
if (SubscriptionPollRequestType.isValidatedRequestType(requestType)) {
switch (SubscriptionPollRequestType.valueOf(requestType)) {
case POLL:
+ final Set pollTopicNames = ((PollPayload) request.getPayload()).getTopicNames();
+ final Set subscribedTopicNames =
+ SubscriptionAgent.consumer()
+ .getTopicNamesSubscribedByConsumer(
+ consumerConfig.getConsumerGroupId(), consumerConfig.getConsumerId());
+ final Set topicNamesToCheck = new HashSet<>(pollTopicNames);
+ topicNamesToCheck.removeIf(topicName -> !subscribedTopicNames.contains(topicName));
+ final TSStatus ownerStatus =
+ SubscriptionAgent.topic().checkTopicOwners(consumerConfig, topicNamesToCheck);
+ if (ownerStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return PipeSubscribePollResp.toTPipeSubscribeResp(ownerStatus, Collections.emptyList());
+ }
events =
handlePipeSubscribePollRequest(
consumerConfig,
@@ -526,11 +550,31 @@ private TPipeSubscribeResp handlePipeSubscribePollInternal(final PipeSubscribePo
request.getProgressByTopic());
break;
case POLL_FILE:
+ final TSStatus tsFileOwnerStatus =
+ SubscriptionAgent.topic()
+ .checkTopicOwner(
+ consumerConfig,
+ ((PollFilePayload) request.getPayload()).getCommitContext().getTopicName());
+ if (tsFileOwnerStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return PipeSubscribePollResp.toTPipeSubscribeResp(
+ tsFileOwnerStatus, Collections.emptyList());
+ }
events =
handlePipeSubscribePollTsFileRequest(
consumerConfig, (PollFilePayload) request.getPayload());
break;
case POLL_TABLETS:
+ final TSStatus tabletsOwnerStatus =
+ SubscriptionAgent.topic()
+ .checkTopicOwner(
+ consumerConfig,
+ ((PollTabletsPayload) request.getPayload())
+ .getCommitContext()
+ .getTopicName());
+ if (tabletsOwnerStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return PipeSubscribePollResp.toTPipeSubscribeResp(
+ tabletsOwnerStatus, Collections.emptyList());
+ }
events =
handlePipeSubscribePollTabletsRequest(
consumerConfig, (PollTabletsPayload) request.getPayload());
@@ -709,6 +753,16 @@ private TPipeSubscribeResp handlePipeSubscribeCommitInternal(final PipeSubscribe
// commit (ack or nack)
final List commitContexts = req.getCommitContexts();
+ final TSStatus ownerStatus =
+ SubscriptionAgent.topic()
+ .checkTopicOwners(
+ consumerConfig,
+ commitContexts.stream()
+ .map(SubscriptionCommitContext::getTopicName)
+ .collect(Collectors.toSet()));
+ if (ownerStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return PipeSubscribeCommitResp.toTPipeSubscribeResp(ownerStatus);
+ }
final boolean nack = req.isNack();
final Set subscribedTopicNames =
SubscriptionAgent.consumer()
@@ -885,6 +939,15 @@ private TPipeSubscribeResp handleSubscriptionSeekInternal(final SubscriptionSeek
final String topicName = req.getTopicName();
final short seekType = req.getSeekType();
+ // Owner fencing: seek mutates the consumption progress, so a stale owner must not be allowed to
+ // move the position after an ownership transfer (otherwise it could corrupt the new owner's
+ // recovery point). This must be guarded just like commit.
+ final TSStatus ownerStatus =
+ SubscriptionAgent.topic().checkTopicOwner(consumerConfig, topicName);
+ if (ownerStatus.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ return PipeSubscribeSeekResp.toTPipeSubscribeResp(ownerStatus);
+ }
+
if (seekType == SubscriptionSeekReq.SEEK_TO_TOPIC_PROGRESS) {
SubscriptionAgent.broker()
.seekToTopicProgress(consumerConfig, topicName, req.getTopicProgress());
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/parser/StatementGeneratorTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/parser/StatementGeneratorTest.java
index b98f34a2484d9..febf29bea1553 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/parser/StatementGeneratorTest.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/parser/StatementGeneratorTest.java
@@ -56,6 +56,7 @@
import org.apache.iotdb.db.queryengine.plan.statement.metadata.DatabaseSchemaStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.DeleteDatabaseStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.DeleteTimeSeriesStatement;
+import org.apache.iotdb.db.queryengine.plan.statement.metadata.subscription.AlterTopicStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.template.BatchActivateTemplateStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.template.CreateSchemaTemplateStatement;
import org.apache.iotdb.db.queryengine.plan.statement.metadata.template.DropSchemaTemplateStatement;
@@ -121,6 +122,20 @@
public class StatementGeneratorTest {
+ @Test
+ public void testAlterTopicStatement() {
+ final Statement statement =
+ StatementGenerator.createStatement(
+ "ALTER TOPIC topic1 WITH ('owner-id'='owner2','owner-epoch'='6')",
+ ZonedDateTime.now().getOffset());
+
+ Assert.assertTrue(statement instanceof AlterTopicStatement);
+ final AlterTopicStatement alterTopicStatement = (AlterTopicStatement) statement;
+ Assert.assertEquals("topic1", alterTopicStatement.getTopicName());
+ Assert.assertEquals("owner2", alterTopicStatement.getTopicAttributes().get("owner-id"));
+ Assert.assertEquals("6", alterTopicStatement.getTopicAttributes().get("owner-epoch"));
+ }
+
@Test
public void testShowDiskUsage() {
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/TopicStatementTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/TopicStatementTest.java
new file mode 100644
index 0000000000000..d457bdbe55f44
--- /dev/null
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/queryengine/plan/relational/sql/parser/TopicStatementTest.java
@@ -0,0 +1,48 @@
+/*
+ * 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.queryengine.plan.relational.sql.parser;
+
+import org.apache.iotdb.commons.queryengine.plan.relational.sql.ast.Statement;
+import org.apache.iotdb.db.queryengine.plan.relational.sql.ast.AlterTopic;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.time.ZoneId;
+
+public class TopicStatementTest {
+
+ private final SqlParser sqlParser = new SqlParser();
+
+ @Test
+ public void testAlterTopicStatement() {
+ final Statement statement =
+ sqlParser.createStatement(
+ "ALTER TOPIC topic1 WITH ('owner-id'='owner2','owner-epoch'='6')",
+ ZoneId.systemDefault(),
+ null);
+
+ Assert.assertTrue(statement instanceof AlterTopic);
+ final AlterTopic alterTopic = (AlterTopic) statement;
+ Assert.assertEquals("topic1", alterTopic.getTopicName());
+ Assert.assertEquals("owner2", alterTopic.getTopicAttributes().get("owner-id"));
+ Assert.assertEquals("6", alterTopic.getTopicAttributes().get("owner-epoch"));
+ }
+}
diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1Test.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1Test.java
index ba0070187e368..0890ae8100aaa 100644
--- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1Test.java
+++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/receiver/SubscriptionReceiverV1Test.java
@@ -20,14 +20,19 @@
package org.apache.iotdb.db.subscription.receiver;
import org.apache.iotdb.commons.subscription.config.SubscriptionConfig;
+import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta;
+import org.apache.iotdb.db.subscription.agent.SubscriptionAgent;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.iotdb.rpc.subscription.config.ConsumerConfig;
import org.apache.iotdb.rpc.subscription.config.ConsumerConstant;
+import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.junit.Assert;
import org.junit.Test;
import java.lang.reflect.Field;
import java.lang.reflect.Method;
+import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
@@ -82,6 +87,77 @@ public void testCalculateConsumerInactivityTimeoutUsesHeartbeatMultiple() throws
invokeCalculateConsumerInactivityTimeoutMs(receiver, createConsumerConfig(5_000L)));
}
+ @Test
+ public void testTopicOwnerFencingStatus() {
+ final String topicName = "topic-" + UUID.randomUUID();
+
+ SubscriptionAgent.topic()
+ .handleSingleTopicMetaChanges(createTopicMeta(topicName, "owner1", 7L));
+ try {
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ SubscriptionAgent.topic()
+ .checkTopicOwner(createConsumerConfig(1_000L, "owner1", 7L), topicName)
+ .getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUBSCRIPTION_OWNER_FENCED.getStatusCode(),
+ SubscriptionAgent.topic()
+ .checkTopicOwner(createConsumerConfig(1_000L, "owner2", 7L), topicName)
+ .getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUBSCRIPTION_OWNER_REQUIRED.getStatusCode(),
+ SubscriptionAgent.topic()
+ .checkTopicOwner(createConsumerConfig(1_000L), topicName)
+ .getCode());
+ } finally {
+ SubscriptionAgent.topic().handleDropTopic(topicName);
+ }
+ }
+
+ @Test
+ public void testOldOwnerFencedAfterNetworkPartitionAndTopicOwnerTransfer() {
+ final String topicName = "topic-" + UUID.randomUUID();
+ final TopicMeta topicMeta = createTopicMeta(topicName, "owner1", 5L);
+ final ConsumerConfig oldOwnerConsumer = createConsumerConfig(1_000L, "owner1", 5L);
+ final ConsumerConfig newOwnerConsumer = createConsumerConfig(1_000L, "owner2", 6L);
+
+ SubscriptionAgent.topic().handleSingleTopicMetaChanges(topicMeta);
+ try {
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ SubscriptionAgent.topic().checkTopicOwner(oldOwnerConsumer, topicName).getCode());
+
+ final TopicMeta transferredTopicMeta = topicMeta.deepCopy();
+ transferredTopicMeta.transferOwner("owner2", 6L);
+ SubscriptionAgent.topic().handleSingleTopicMetaChanges(transferredTopicMeta);
+
+ Assert.assertEquals(
+ TSStatusCode.SUBSCRIPTION_OWNER_FENCED.getStatusCode(),
+ SubscriptionAgent.topic().checkTopicOwner(oldOwnerConsumer, topicName).getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUBSCRIPTION_OWNER_FENCED.getStatusCode(),
+ SubscriptionAgent.topic()
+ .checkTopicOwners(oldOwnerConsumer, Collections.singleton(topicName))
+ .getCode());
+ Assert.assertNotNull(
+ SubscriptionAgent.topic()
+ .handleSingleTopicMetaChanges(createTopicMeta(topicName, "owner1", 5L)));
+ Assert.assertEquals(
+ TSStatusCode.SUBSCRIPTION_OWNER_FENCED.getStatusCode(),
+ SubscriptionAgent.topic().checkTopicOwner(oldOwnerConsumer, topicName).getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ SubscriptionAgent.topic().checkTopicOwner(newOwnerConsumer, topicName).getCode());
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ SubscriptionAgent.topic()
+ .checkTopicOwners(newOwnerConsumer, Collections.singleton(topicName))
+ .getCode());
+ } finally {
+ SubscriptionAgent.topic().handleDropTopic(topicName);
+ }
+ }
+
private long invokeCalculateConsumerInactivityTimeoutMs(
final SubscriptionReceiverV1 receiver, final ConsumerConfig consumerConfig) throws Exception {
final Method method =
@@ -91,11 +167,30 @@ private long invokeCalculateConsumerInactivityTimeoutMs(
return (long) method.invoke(receiver, consumerConfig);
}
+ private TopicMeta createTopicMeta(
+ final String topicName, final String ownerId, final long ownerEpoch) {
+ final Map topicAttributes = new HashMap<>();
+ topicAttributes.put(TopicConstant.OWNER_ID_KEY, ownerId);
+ topicAttributes.put(TopicConstant.OWNER_EPOCH_KEY, String.valueOf(ownerEpoch));
+ return new TopicMeta(topicName, 1, topicAttributes);
+ }
+
private ConsumerConfig createConsumerConfig(final long heartbeatIntervalMs) {
+ return createConsumerConfig(heartbeatIntervalMs, null, null);
+ }
+
+ private ConsumerConfig createConsumerConfig(
+ final long heartbeatIntervalMs, final String ownerId, final Long ownerEpoch) {
final Map attributes = new HashMap<>();
attributes.put(ConsumerConstant.CONSUMER_ID_KEY, "consumer-" + UUID.randomUUID());
attributes.put(ConsumerConstant.CONSUMER_GROUP_ID_KEY, "group-" + UUID.randomUUID());
attributes.put(ConsumerConstant.HEARTBEAT_INTERVAL_MS_KEY, String.valueOf(heartbeatIntervalMs));
+ if (ownerId != null) {
+ attributes.put(ConsumerConstant.OWNER_ID_KEY, ownerId);
+ }
+ if (ownerEpoch != null) {
+ attributes.put(ConsumerConstant.OWNER_EPOCH_KEY, String.valueOf(ownerEpoch));
+ }
return new ConsumerConfig(attributes);
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
index 252d00b5a4fca..40d801f74e1f5 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonConfig.java
@@ -399,6 +399,12 @@ public class CommonConfig {
private long subscriptionMetaSyncerInitialSyncDelayMinutes = 3;
private long subscriptionMetaSyncerSyncIntervalMinutes = 3;
+ // Minimum allowed owner-lease-duration-ms accepted when creating/altering a topic owner. The
+ // lease is renewed by an independent ~5s heartbeat, so the duration must stay well above the
+ // heartbeat interval (times the tolerated misses plus propagation) to avoid falsely fencing a
+ // healthy owner; this floor enforces that invariant at admission time. Default: 1 minute.
+ private long subscriptionOwnerLeaseDurationMsMin = 60_000L;
+
private int subscriptionConsensusBatchMaxDelayInMs = 50;
private long subscriptionConsensusBatchMaxSizeInBytes = 8 * MB;
private int subscriptionConsensusBatchMaxTabletCount = 64;
@@ -2759,6 +2765,14 @@ public void setSubscriptionMetaSyncerSyncIntervalMinutes(
this.subscriptionMetaSyncerSyncIntervalMinutes = subscriptionMetaSyncerSyncIntervalMinutes;
}
+ public long getSubscriptionOwnerLeaseDurationMsMin() {
+ return subscriptionOwnerLeaseDurationMsMin;
+ }
+
+ public void setSubscriptionOwnerLeaseDurationMsMin(long subscriptionOwnerLeaseDurationMsMin) {
+ this.subscriptionOwnerLeaseDurationMsMin = subscriptionOwnerLeaseDurationMsMin;
+ }
+
public String getSchemaEngineMode() {
return schemaEngineMode;
}
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
index fa853f61f47a8..964bc60b823d0 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/conf/CommonDescriptor.java
@@ -455,6 +455,12 @@ private void loadSubscriptionProps(TrimProperties properties) {
"subscription_meta_syncer_sync_interval_minutes",
String.valueOf(config.getSubscriptionMetaSyncerSyncIntervalMinutes()))));
+ config.setSubscriptionOwnerLeaseDurationMsMin(
+ Long.parseLong(
+ properties.getProperty(
+ "subscription_owner_lease_duration_ms_min",
+ String.valueOf(config.getSubscriptionOwnerLeaseDurationMsMin()))));
+
config.setSubscriptionConsensusBatchMaxDelayInMs(
Integer.parseInt(
properties.getProperty(
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/config/SubscriptionConfig.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/config/SubscriptionConfig.java
index 2fad6cb88da7c..156397e7c1d46 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/config/SubscriptionConfig.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/config/SubscriptionConfig.java
@@ -142,6 +142,10 @@ public long getSubscriptionMetaSyncerSyncIntervalMinutes() {
return COMMON_CONFIG.getSubscriptionMetaSyncerSyncIntervalMinutes();
}
+ public long getSubscriptionOwnerLeaseDurationMsMin() {
+ return COMMON_CONFIG.getSubscriptionOwnerLeaseDurationMsMin();
+ }
+
// Consensus subscription batching parameters
public int getSubscriptionConsensusBatchMaxDelayInMs() {
return COMMON_CONFIG.getSubscriptionConsensusBatchMaxDelayInMs();
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/topic/TopicMeta.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/topic/TopicMeta.java
index badb77d6f486d..c44692f63588d 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/topic/TopicMeta.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/topic/TopicMeta.java
@@ -24,6 +24,8 @@
import org.apache.iotdb.commons.pipe.datastructure.visibility.VisibilityUtils;
import org.apache.iotdb.commons.utils.TestOnly;
import org.apache.iotdb.rpc.subscription.config.TopicConfig;
+import org.apache.iotdb.rpc.subscription.config.TopicConstant;
+import org.apache.iotdb.rpc.subscription.i18n.SubscriptionMessages;
import org.apache.tsfile.utils.PublicBAOS;
import org.apache.tsfile.utils.ReadWriteIOUtils;
@@ -45,11 +47,24 @@ public class TopicMeta {
private long creationTime; // unit in ms
private TopicConfig config;
+ private String ownerId;
+ private long ownerEpoch;
+ // Highest owner epoch ever granted for this topic; retained across clearOwner. Mirrors the
+ // MAX_OWNER_EPOCH_KEY attribute (which carries it through serialization). Used to guarantee owner
+ // epoch is globally monotonic and never reused.
+ private long maxOwnerEpoch;
+ private long ownerLastTransferTimeMs;
+ private Long ownerLeaseDurationMs;
+ private Long ownerLeaseExpireTimeMs;
+
// TODO: remove this variable later
private Set subscribedConsumerGroupIds; // unused now
private TopicMeta() {
this.config = new TopicConfig(new HashMap<>());
+ this.ownerEpoch = -1L;
+ this.maxOwnerEpoch = -1L;
+ this.ownerLastTransferTimeMs = -1L;
this.subscribedConsumerGroupIds = new HashSet<>();
}
@@ -59,6 +74,10 @@ public TopicMeta(
this.topicName = topicName;
this.creationTime = creationTime;
this.config = new TopicConfig(topicAttributes);
+ this.ownerEpoch = -1L;
+ this.maxOwnerEpoch = -1L;
+ this.ownerLastTransferTimeMs = -1L;
+ initOwnerFromTopicAttributes(topicAttributes);
this.subscribedConsumerGroupIds = new HashSet<>();
}
@@ -68,7 +87,29 @@ public TopicMeta deepCopy() {
copied.topicName = topicName;
copied.creationTime = creationTime;
copied.config = new TopicConfig(new HashMap<>(config.getAttribute()));
+ copied.ownerId = ownerId;
+ copied.ownerEpoch = ownerEpoch;
+ copied.maxOwnerEpoch = maxOwnerEpoch;
+ copied.ownerLastTransferTimeMs = ownerLastTransferTimeMs;
+ copied.ownerLeaseDurationMs = ownerLeaseDurationMs;
+ copied.ownerLeaseExpireTimeMs = ownerLeaseExpireTimeMs;
+
+ copied.subscribedConsumerGroupIds = new HashSet<>(subscribedConsumerGroupIds);
+ return copied;
+ }
+
+ public TopicMeta deepCopyWithUpdatedAttributes(final Map updatedAttributes) {
+ final Map copiedAttributes = new HashMap<>(config.getAttribute());
+ if (Objects.nonNull(updatedAttributes)) {
+ copiedAttributes.putAll(updatedAttributes);
+ if ((updatedAttributes.containsKey(TopicConstant.OWNER_ID_KEY)
+ || updatedAttributes.containsKey(TopicConstant.OWNER_EPOCH_KEY))
+ && !updatedAttributes.containsKey(TopicConstant.OWNER_LEASE_DURATION_MS_KEY)) {
+ copiedAttributes.remove(TopicConstant.OWNER_LEASE_DURATION_MS_KEY);
+ }
+ }
+ final TopicMeta copied = new TopicMeta(topicName, creationTime, copiedAttributes);
copied.subscribedConsumerGroupIds = new HashSet<>(subscribedConsumerGroupIds);
return copied;
}
@@ -85,6 +126,125 @@ public TopicConfig getConfig() {
return config;
}
+ public boolean isOwnerFencingEnabled() {
+ return Objects.nonNull(ownerId) && ownerEpoch >= 0;
+ }
+
+ public String getOwnerId() {
+ return ownerId;
+ }
+
+ public long getOwnerEpoch() {
+ return ownerEpoch;
+ }
+
+ public long getMaxOwnerEpoch() {
+ return maxOwnerEpoch;
+ }
+
+ public long getOwnerLastTransferTimeMs() {
+ return ownerLastTransferTimeMs;
+ }
+
+ public Long getOwnerLeaseExpireTimeMs() {
+ return ownerLeaseExpireTimeMs;
+ }
+
+ public Long getOwnerLeaseDurationMs() {
+ return ownerLeaseDurationMs;
+ }
+
+ public void transferOwner(final String ownerId, final long ownerEpoch) {
+ transferOwner(ownerId, ownerEpoch, null);
+ }
+
+ public void transferOwner(
+ final String ownerId, final long ownerEpoch, final Long ownerLeaseDurationMs) {
+ if (Objects.isNull(ownerId) || ownerId.isEmpty()) {
+ throw new IllegalArgumentException(SubscriptionMessages.OWNER_ID_SHOULD_NOT_BE_EMPTY);
+ }
+ if (ownerEpoch < 0) {
+ throw new IllegalArgumentException(SubscriptionMessages.OWNER_EPOCH_SHOULD_NOT_BE_NEGATIVE);
+ }
+ if (isOwnerFencingEnabled() && ownerEpoch <= this.ownerEpoch) {
+ throw new IllegalArgumentException(
+ String.format(
+ SubscriptionMessages.OWNER_EPOCH_SHOULD_INCREASE_MONOTONICALLY,
+ this.ownerEpoch,
+ ownerEpoch));
+ }
+ if (Objects.nonNull(ownerLeaseDurationMs) && ownerLeaseDurationMs <= 0) {
+ throw new IllegalArgumentException(
+ String.format(
+ SubscriptionMessages.OWNER_LEASE_DURATION_SHOULD_BE_POSITIVE,
+ topicName,
+ ownerLeaseDurationMs));
+ }
+
+ this.ownerId = ownerId;
+ this.ownerEpoch = ownerEpoch;
+ this.ownerLeaseDurationMs = ownerLeaseDurationMs;
+ // The lease expire time is DataNode-local runtime state derived from the owner-lease heartbeat;
+ // a freshly (re)assigned owner has no lease until the first heartbeat establishes one.
+ this.ownerLeaseExpireTimeMs = null;
+ this.ownerLastTransferTimeMs = System.currentTimeMillis();
+
+ config.getAttribute().put(TopicConstant.OWNER_ID_KEY, ownerId);
+ config.getAttribute().put(TopicConstant.OWNER_EPOCH_KEY, String.valueOf(ownerEpoch));
+ if (ownerEpoch > maxOwnerEpoch) {
+ maxOwnerEpoch = ownerEpoch;
+ }
+ config.getAttribute().put(TopicConstant.MAX_OWNER_EPOCH_KEY, String.valueOf(maxOwnerEpoch));
+ if (Objects.nonNull(ownerLeaseDurationMs)) {
+ config
+ .getAttribute()
+ .put(TopicConstant.OWNER_LEASE_DURATION_MS_KEY, String.valueOf(ownerLeaseDurationMs));
+ } else {
+ config.getAttribute().remove(TopicConstant.OWNER_LEASE_DURATION_MS_KEY);
+ }
+ }
+
+ public void clearOwner() {
+ ownerId = null;
+ ownerEpoch = -1L;
+ ownerLastTransferTimeMs = -1L;
+ ownerLeaseDurationMs = null;
+ ownerLeaseExpireTimeMs = null;
+ config.getAttribute().remove(TopicConstant.OWNER_ID_KEY);
+ config.getAttribute().remove(TopicConstant.OWNER_EPOCH_KEY);
+ config.getAttribute().remove(TopicConstant.OWNER_LEASE_DURATION_MS_KEY);
+ // Intentionally retain maxOwnerEpoch and MAX_OWNER_EPOCH_KEY so that a later re-enable cannot
+ // reuse an epoch <= any epoch ever granted (global monotonicity survives clear).
+ }
+
+ public boolean matchesOwner(final String requestOwnerId, final Long requestOwnerEpoch) {
+ return !isOwnerFencingEnabled()
+ || (Objects.equals(ownerId, requestOwnerId)
+ && Objects.equals(ownerEpoch, requestOwnerEpoch)
+ && (Objects.isNull(ownerLeaseExpireTimeMs)
+ || System.currentTimeMillis() <= ownerLeaseExpireTimeMs));
+ }
+
+ /**
+ * DataNode side: apply a lease renewal pushed by ConfigNode via the dedicated subscription owner
+ * heartbeat. The pushed value is a relative remaining duration; it is converted to a local
+ * absolute expire time using this node's own clock, so no absolute timestamp is ever compared
+ * across ConfigNode and DataNode clocks. This expire time is purely DataNode-local runtime state
+ * (not a topic attribute, not serialized, not part of equality) so it never propagates back. Only
+ * refreshes when the pushed {@code (ownerId, ownerEpoch)} matches the current owner; stale or
+ * mismatched pushes are ignored (owner identity/epoch changes arrive via the topic-meta push
+ * path, not the lease heartbeat).
+ */
+ public void applyOwnerLeaseFromHeartbeat(
+ final String ownerId, final long ownerEpoch, final long leaseRemainingMs) {
+ if (!isOwnerFencingEnabled()
+ || !Objects.equals(this.ownerId, ownerId)
+ || this.ownerEpoch != ownerEpoch) {
+ return;
+ }
+ this.ownerLeaseExpireTimeMs = System.currentTimeMillis() + leaseRemainingMs;
+ }
+
/**
* @return true if the consumer group did not already subscribe this topic
*/
@@ -156,6 +316,8 @@ public static TopicMeta deserialize(final InputStream inputStream) throws IOExce
topicMeta.subscribedConsumerGroupIds.add(ReadWriteIOUtils.readString(inputStream));
}
+ topicMeta.initOwnerFromTopicAttributes(topicMeta.config.getAttribute());
+
return topicMeta;
}
@@ -177,9 +339,95 @@ public static TopicMeta deserialize(final ByteBuffer byteBuffer) {
topicMeta.subscribedConsumerGroupIds.add(ReadWriteIOUtils.readString(byteBuffer));
}
+ topicMeta.initOwnerFromTopicAttributes(topicMeta.config.getAttribute());
+
return topicMeta;
}
+ public static void validateOwnerProgression(
+ final TopicMeta currentTopicMeta, final TopicMeta updatedTopicMeta) {
+ if (Objects.isNull(currentTopicMeta) || Objects.isNull(updatedTopicMeta)) {
+ return;
+ }
+
+ // Global epoch monotonicity (never reused): a transfer to a different owner must use an epoch
+ // strictly greater than any epoch ever granted for this topic. The baseline (maxOwnerEpoch) is
+ // retained across clearOwner, so this also blocks epoch reuse after a clear -> re-enable cycle.
+ // Same-owner updates (e.g. attribute / lease changes) are unaffected.
+ if (updatedTopicMeta.isOwnerFencingEnabled()
+ && !Objects.equals(updatedTopicMeta.getOwnerId(), currentTopicMeta.getOwnerId())
+ && updatedTopicMeta.getOwnerEpoch() <= currentTopicMeta.getMaxOwnerEpoch()) {
+ throw new IllegalArgumentException(
+ String.format(
+ SubscriptionMessages.OWNER_EPOCH_SHOULD_NEVER_BE_REUSED,
+ currentTopicMeta.getTopicName(),
+ currentTopicMeta.getMaxOwnerEpoch(),
+ updatedTopicMeta.getOwnerId(),
+ updatedTopicMeta.getOwnerEpoch()));
+ }
+
+ if (!currentTopicMeta.isOwnerFencingEnabled()) {
+ return;
+ }
+
+ if (!updatedTopicMeta.isOwnerFencingEnabled()) {
+ throw new IllegalArgumentException(
+ String.format(
+ SubscriptionMessages.OWNER_SHOULD_NOT_BE_CLEARED_BY_STALE_META,
+ currentTopicMeta.getTopicName(),
+ currentTopicMeta.getOwnerId(),
+ currentTopicMeta.getOwnerEpoch()));
+ }
+
+ final boolean epochRollback =
+ updatedTopicMeta.getOwnerEpoch() < currentTopicMeta.getOwnerEpoch();
+ final boolean sameEpochOwnerChanged =
+ updatedTopicMeta.getOwnerEpoch() == currentTopicMeta.getOwnerEpoch()
+ && !Objects.equals(updatedTopicMeta.getOwnerId(), currentTopicMeta.getOwnerId());
+ if (epochRollback || sameEpochOwnerChanged) {
+ throw new IllegalArgumentException(
+ String.format(
+ SubscriptionMessages.OWNER_EPOCH_SHOULD_NOT_ROLL_BACK,
+ currentTopicMeta.getTopicName(),
+ currentTopicMeta.getOwnerId(),
+ currentTopicMeta.getOwnerEpoch(),
+ updatedTopicMeta.getOwnerId(),
+ updatedTopicMeta.getOwnerEpoch()));
+ }
+ }
+
+ private void initOwnerFromTopicAttributes(final Map topicAttributes) {
+ final TopicConfig topicConfig = new TopicConfig(topicAttributes);
+
+ // Restore the global epoch baseline first; it persists even when the owner has been cleared.
+ final Long configuredMaxOwnerEpoch = topicConfig.getLong(TopicConstant.MAX_OWNER_EPOCH_KEY);
+ if (Objects.nonNull(configuredMaxOwnerEpoch)) {
+ this.maxOwnerEpoch = configuredMaxOwnerEpoch;
+ }
+
+ final String configuredOwnerId = topicConfig.getString(TopicConstant.OWNER_ID_KEY);
+ if (Objects.isNull(configuredOwnerId)) {
+ return;
+ }
+
+ final Long configuredOwnerEpoch = topicConfig.getLong(TopicConstant.OWNER_EPOCH_KEY);
+ if (Objects.isNull(configuredOwnerEpoch)) {
+ throw new IllegalArgumentException(
+ String.format(
+ SubscriptionMessages.OWNER_EPOCH_SHOULD_BE_SET_WHEN_OWNER_ID_SET,
+ TopicConstant.OWNER_ID_KEY));
+ }
+ final Long ownerLeaseDurationMs =
+ topicConfig.getLong(TopicConstant.OWNER_LEASE_DURATION_MS_KEY);
+ if (Objects.nonNull(ownerLeaseDurationMs) && ownerLeaseDurationMs <= 0) {
+ throw new IllegalArgumentException(
+ String.format(
+ SubscriptionMessages.OWNER_LEASE_DURATION_SHOULD_BE_POSITIVE_WHEN_SET,
+ TopicConstant.OWNER_LEASE_DURATION_MS_KEY));
+ }
+ transferOwner(configuredOwnerId, configuredOwnerEpoch, ownerLeaseDurationMs);
+ }
+
/////////////////////////////// utilities ///////////////////////////////
public Map generateExtractorAttributes(final String username) {
@@ -257,12 +505,17 @@ public boolean equals(final Object obj) {
final TopicMeta that = (TopicMeta) obj;
return creationTime == that.creationTime
&& Objects.equals(topicName, that.topicName)
- && Objects.equals(config, that.config);
+ && Objects.equals(config, that.config)
+ && Objects.equals(ownerId, that.ownerId)
+ && ownerEpoch == that.ownerEpoch
+ && maxOwnerEpoch == that.maxOwnerEpoch
+ && Objects.equals(ownerLeaseDurationMs, that.ownerLeaseDurationMs);
}
@Override
public int hashCode() {
- return Objects.hash(topicName, creationTime, config);
+ return Objects.hash(
+ topicName, creationTime, config, ownerId, ownerEpoch, maxOwnerEpoch, ownerLeaseDurationMs);
}
@Override
@@ -274,6 +527,16 @@ public String toString() {
+ creationTime
+ ", config="
+ config
+ + ", ownerId='"
+ + ownerId
+ + "', ownerEpoch="
+ + ownerEpoch
+ + ", ownerLastTransferTimeMs="
+ + ownerLastTransferTimeMs
+ + ", ownerLeaseDurationMs="
+ + ownerLeaseDurationMs
+ + ", ownerLeaseExpireTimeMs="
+ + ownerLeaseExpireTimeMs
+ '}';
}
}
diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/topic/TopicDeSerTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/topic/TopicDeSerTest.java
index d9c280e14938c..fa47c7241b521 100644
--- a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/topic/TopicDeSerTest.java
+++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/topic/TopicDeSerTest.java
@@ -20,10 +20,13 @@
package org.apache.iotdb.commons.subscription.topic;
import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta;
+import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.junit.Assert;
import org.junit.Test;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
@@ -55,6 +58,166 @@ public void test() throws IOException {
topicMeta.getSubscribedConsumerGroupIds(), topicMeta2.getSubscribedConsumerGroupIds());
}
+ @Test
+ public void testTopicOwnerDeSer() throws IOException {
+ Map topicAttributes = new HashMap<>();
+ topicAttributes.put(TopicConstant.OWNER_ID_KEY, "owner1");
+ topicAttributes.put(TopicConstant.OWNER_EPOCH_KEY, "5");
+
+ TopicMeta topicMeta = new TopicMeta("test_topic", 1, topicAttributes);
+
+ Assert.assertTrue(topicMeta.isOwnerFencingEnabled());
+ Assert.assertEquals("owner1", topicMeta.getOwnerId());
+ Assert.assertEquals(5L, topicMeta.getOwnerEpoch());
+ Assert.assertTrue(topicMeta.matchesOwner("owner1", 5L));
+ Assert.assertFalse(topicMeta.matchesOwner("owner2", 5L));
+ Assert.assertFalse(topicMeta.matchesOwner("owner1", 4L));
+
+ TopicMeta topicMeta1 = TopicMeta.deserialize(topicMeta.serialize());
+ TopicMeta topicMeta2 = topicMeta1.deepCopy();
+
+ Assert.assertEquals(topicMeta, topicMeta1);
+ Assert.assertEquals(topicMeta, topicMeta2);
+ Assert.assertEquals(topicMeta.getOwnerId(), topicMeta2.getOwnerId());
+ Assert.assertEquals(topicMeta.getOwnerEpoch(), topicMeta2.getOwnerEpoch());
+ Assert.assertEquals(topicMeta.getOwnerLeaseDurationMs(), topicMeta2.getOwnerLeaseDurationMs());
+ Assert.assertEquals(
+ topicMeta.getOwnerLeaseExpireTimeMs(), topicMeta2.getOwnerLeaseExpireTimeMs());
+
+ topicMeta.transferOwner("owner2", 6L, 100L);
+ Assert.assertEquals("owner2", topicMeta.getOwnerId());
+ Assert.assertEquals(6L, topicMeta.getOwnerEpoch());
+ Assert.assertEquals("owner2", topicMeta.getConfig().getString(TopicConstant.OWNER_ID_KEY));
+ Assert.assertEquals(
+ 6L, topicMeta.getConfig().getLong(TopicConstant.OWNER_EPOCH_KEY).longValue());
+ Assert.assertEquals(
+ 100L, topicMeta.getConfig().getLong(TopicConstant.OWNER_LEASE_DURATION_MS_KEY).longValue());
+ Assert.assertEquals(100L, topicMeta.getOwnerLeaseDurationMs().longValue());
+ // A freshly transferred owner has no lease expiry until the first owner-lease heartbeat.
+ Assert.assertNull(topicMeta.getOwnerLeaseExpireTimeMs());
+
+ topicMeta.clearOwner();
+ Assert.assertFalse(topicMeta.isOwnerFencingEnabled());
+ Assert.assertFalse(topicMeta.getConfig().hasAttribute(TopicConstant.OWNER_ID_KEY));
+ Assert.assertFalse(topicMeta.getConfig().hasAttribute(TopicConstant.OWNER_EPOCH_KEY));
+ Assert.assertFalse(
+ topicMeta.getConfig().hasAttribute(TopicConstant.OWNER_LEASE_DURATION_MS_KEY));
+ }
+
+ @Test
+ public void testDeepCopyWithUpdatedAttributesTransfersOwnerAndPreservesExistingConfig() {
+ final Map topicAttributes = new HashMap<>();
+ topicAttributes.put(TopicConstant.PATH_KEY, "root.sg.**");
+ topicAttributes.put(TopicConstant.OWNER_ID_KEY, "owner1");
+ topicAttributes.put(TopicConstant.OWNER_EPOCH_KEY, "5");
+ final TopicMeta topicMeta = new TopicMeta("test_topic", 1, topicAttributes);
+ topicMeta.addSubscribedConsumerGroup("group1");
+
+ final Map updatedAttributes = new HashMap<>();
+ updatedAttributes.put(TopicConstant.OWNER_ID_KEY, "owner2");
+ updatedAttributes.put(TopicConstant.OWNER_EPOCH_KEY, "6");
+ final TopicMeta updatedTopicMeta = topicMeta.deepCopyWithUpdatedAttributes(updatedAttributes);
+
+ Assert.assertEquals(
+ "root.sg.**", updatedTopicMeta.getConfig().getString(TopicConstant.PATH_KEY));
+ Assert.assertEquals("owner2", updatedTopicMeta.getOwnerId());
+ Assert.assertEquals(6L, updatedTopicMeta.getOwnerEpoch());
+ Assert.assertTrue(updatedTopicMeta.isSubscribedByConsumerGroup("group1"));
+ Assert.assertEquals("owner1", topicMeta.getOwnerId());
+ Assert.assertEquals(5L, topicMeta.getOwnerEpoch());
+ }
+
+ @Test
+ public void testDeepCopyWithUpdatedAttributesTransfersOwnerAndClearsExistingLease() {
+ final Map topicAttributes = new HashMap<>();
+ topicAttributes.put(TopicConstant.OWNER_ID_KEY, "owner1");
+ topicAttributes.put(TopicConstant.OWNER_EPOCH_KEY, "5");
+ topicAttributes.put(TopicConstant.OWNER_LEASE_DURATION_MS_KEY, "60000");
+ final TopicMeta topicMeta = new TopicMeta("test_topic", 1, topicAttributes);
+
+ final Map updatedAttributes = new HashMap<>();
+ updatedAttributes.put(TopicConstant.OWNER_ID_KEY, "owner2");
+ updatedAttributes.put(TopicConstant.OWNER_EPOCH_KEY, "6");
+ final TopicMeta updatedTopicMeta = topicMeta.deepCopyWithUpdatedAttributes(updatedAttributes);
+
+ Assert.assertEquals("owner2", updatedTopicMeta.getOwnerId());
+ Assert.assertEquals(6L, updatedTopicMeta.getOwnerEpoch());
+ Assert.assertNull(updatedTopicMeta.getOwnerLeaseDurationMs());
+ Assert.assertNull(updatedTopicMeta.getOwnerLeaseExpireTimeMs());
+ Assert.assertFalse(
+ updatedTopicMeta.getConfig().hasAttribute(TopicConstant.OWNER_LEASE_DURATION_MS_KEY));
+ }
+
+ @Test
+ public void testOwnerLeaseExpireIsDataNodeLocalRuntimeOnly() throws IOException {
+ final Map topicAttributes = new HashMap<>();
+ topicAttributes.put(TopicConstant.OWNER_ID_KEY, "owner1");
+ topicAttributes.put(TopicConstant.OWNER_EPOCH_KEY, "5");
+ topicAttributes.put(TopicConstant.OWNER_LEASE_DURATION_MS_KEY, "60000");
+ final TopicMeta topicMeta = new TopicMeta("test_topic", 1, topicAttributes);
+
+ // Construction (ConfigNode side) never derives an absolute expiry; that is established only by
+ // the DataNode-local owner-lease heartbeat.
+ Assert.assertNull(topicMeta.getOwnerLeaseExpireTimeMs());
+ Assert.assertTrue(topicMeta.matchesOwner("owner1", 5L));
+
+ // Apply a heartbeat with a relative remaining duration; expiry is now set locally and the owner
+ // still matches before the lease drains.
+ topicMeta.applyOwnerLeaseFromHeartbeat("owner1", 5L, 60000L);
+ Assert.assertNotNull(topicMeta.getOwnerLeaseExpireTimeMs());
+ Assert.assertTrue(topicMeta.matchesOwner("owner1", 5L));
+
+ // The locally-derived expiry is runtime-only: it is not a topic attribute, does not affect
+ // equality, and does not survive serialization.
+ final TopicMeta roundTrip = TopicMeta.deserialize(topicMeta.serialize());
+ Assert.assertNull(roundTrip.getOwnerLeaseExpireTimeMs());
+ Assert.assertEquals(topicMeta, roundTrip);
+
+ // A stale/mismatched heartbeat is ignored.
+ topicMeta.applyOwnerLeaseFromHeartbeat("owner2", 5L, 60000L);
+ topicMeta.applyOwnerLeaseFromHeartbeat("owner1", 4L, 60000L);
+ Assert.assertTrue(topicMeta.matchesOwner("owner1", 5L));
+
+ // An elapsed lease causes the current owner to stop matching (DataNode fencing).
+ topicMeta.applyOwnerLeaseFromHeartbeat("owner1", 5L, -1L);
+ Assert.assertFalse(topicMeta.matchesOwner("owner1", 5L));
+ }
+
+ @Test
+ public void testOwnerLeaseDurationGeneratesExpireTime() {
+ final long ownerLeaseDurationMs = 60000;
+ final Map topicAttributes = new HashMap<>();
+ topicAttributes.put(TopicConstant.OWNER_ID_KEY, "owner1");
+ topicAttributes.put(TopicConstant.OWNER_EPOCH_KEY, "5");
+ topicAttributes.put(
+ TopicConstant.OWNER_LEASE_DURATION_MS_KEY, String.valueOf(ownerLeaseDurationMs));
+
+ final TopicMeta topicMeta = new TopicMeta("test_topic", 1, topicAttributes);
+
+ // The lease duration is persisted, but no absolute expiry is derived at construction time: the
+ // expiry is established only by the DataNode-local owner-lease heartbeat.
+ Assert.assertEquals(ownerLeaseDurationMs, topicMeta.getOwnerLeaseDurationMs().longValue());
+ Assert.assertEquals(
+ ownerLeaseDurationMs,
+ topicMeta.getConfig().getLong(TopicConstant.OWNER_LEASE_DURATION_MS_KEY).longValue());
+ Assert.assertNull(topicMeta.getOwnerLeaseExpireTimeMs());
+ }
+
+ @Test
+ public void testSequentialTopicMetaDeserializeDoesNotConsumeNextTopic() throws IOException {
+ final TopicMeta firstTopicMeta = new TopicMeta("first_topic", 1, new HashMap<>());
+ final TopicMeta secondTopicMeta = new TopicMeta("second_topic", 2, new HashMap<>());
+
+ final ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
+ firstTopicMeta.serialize(outputStream);
+ secondTopicMeta.serialize(outputStream);
+
+ final ByteArrayInputStream inputStream = new ByteArrayInputStream(outputStream.toByteArray());
+ Assert.assertEquals(firstTopicMeta, TopicMeta.deserialize(inputStream));
+ Assert.assertEquals(secondTopicMeta, TopicMeta.deserialize(inputStream));
+ Assert.assertEquals(0, inputStream.available());
+ }
+
@Test
public void testGenerateExtractorAttributesWithEncryptedPassword() {
final TopicMeta topicMeta = new TopicMeta("test_topic", 1, new HashMap<>());
@@ -65,4 +228,41 @@ public void testGenerateExtractorAttributesWithEncryptedPassword() {
Assert.assertEquals("test_user", extractorAttributes.get("username"));
Assert.assertEquals("encrypted-password", extractorAttributes.get("password"));
}
+
+ @Test
+ public void testOwnerEpochNeverReusedAcrossClearAndSerialization() throws IOException {
+ final Map topicAttributes = new HashMap<>();
+ topicAttributes.put(TopicConstant.OWNER_ID_KEY, "owner1");
+ topicAttributes.put(TopicConstant.OWNER_EPOCH_KEY, "6");
+ final TopicMeta topicMeta = new TopicMeta("test_topic", 1, topicAttributes);
+ Assert.assertEquals(6L, topicMeta.getMaxOwnerEpoch());
+
+ // Clearing the owner retains the global epoch baseline.
+ topicMeta.clearOwner();
+ Assert.assertFalse(topicMeta.isOwnerFencingEnabled());
+ Assert.assertEquals(6L, topicMeta.getMaxOwnerEpoch());
+
+ // The baseline survives serialization even though the owner is cleared.
+ final TopicMeta deserialized = TopicMeta.deserialize(topicMeta.serialize());
+ Assert.assertFalse(deserialized.isOwnerFencingEnabled());
+ Assert.assertEquals(6L, deserialized.getMaxOwnerEpoch());
+
+ // Re-enabling with an epoch <= the historical max (reuse) is rejected.
+ final Map reuse = new HashMap<>();
+ reuse.put(TopicConstant.OWNER_ID_KEY, "owner2");
+ reuse.put(TopicConstant.OWNER_EPOCH_KEY, "6");
+ final TopicMeta reuseMeta = deserialized.deepCopyWithUpdatedAttributes(reuse);
+ Assert.assertThrows(
+ IllegalArgumentException.class,
+ () -> TopicMeta.validateOwnerProgression(deserialized, reuseMeta));
+
+ // Re-enabling with an epoch strictly greater than the historical max is allowed.
+ final Map next = new HashMap<>();
+ next.put(TopicConstant.OWNER_ID_KEY, "owner2");
+ next.put(TopicConstant.OWNER_EPOCH_KEY, "7");
+ final TopicMeta nextMeta = deserialized.deepCopyWithUpdatedAttributes(next);
+ TopicMeta.validateOwnerProgression(deserialized, nextMeta);
+ Assert.assertEquals(7L, nextMeta.getOwnerEpoch());
+ Assert.assertEquals(7L, nextMeta.getMaxOwnerEpoch());
+ }
}
diff --git a/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4 b/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4
index 93dfa8bb1f4f1..55f965ba4abc9 100644
--- a/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4
+++ b/iotdb-core/relational-grammar/src/main/antlr4/org/apache/iotdb/db/relational/grammar/sql/RelationalSql.g4
@@ -108,6 +108,7 @@ statement
// Subscription Statement
| createTopicStatement
+ | alterTopicStatement
| dropTopicStatement
| showTopicsStatement
| showSubscriptionsStatement
@@ -537,6 +538,10 @@ createTopicStatement
: CREATE TOPIC (IF NOT EXISTS)? topicName=identifier topicAttributesClause?
;
+alterTopicStatement
+ : ALTER TOPIC topicName=identifier topicAttributesClause
+ ;
+
topicAttributesClause
: WITH '(' topicAttributeClause (',' topicAttributeClause)* ')'
;
diff --git a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
index 26600f4ea9b47..b0b20ee4ac701 100644
--- a/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
+++ b/iotdb-protocol/thrift-confignode/src/main/thrift/confignode.thrift
@@ -1942,6 +1942,9 @@ service IConfigNodeRPCService {
/** Create Topic */
common.TSStatus createTopic(TCreateTopicReq req)
+ /** Alter Topic */
+ common.TSStatus alterTopic(TAlterTopicReq req)
+
/** Drop Topic */
common.TSStatus dropTopic(string topicName)
diff --git a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
index 2f3703ec5fcfd..11de2e4cba232 100644
--- a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
+++ b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
@@ -582,6 +582,21 @@ struct TPushTopicMetaRespExceptionMessage {
3: required i64 timeStamp
}
+// Dedicated subscription owner-lease heartbeat (CN -> DN), decoupled from the node heartbeat.
+// Carries a relative remaining duration; the DataNode derives the local expire time on its own
+// clock (leaseExpireAt = localNow + leaseRemainingMs), so no absolute timestamp is compared across
+// ConfigNode and DataNode clocks.
+struct TTopicOwnerLeaseEntry {
+ 1: required string topicName
+ 2: required string ownerId
+ 3: required i64 ownerEpoch
+ 4: required i64 leaseRemainingMs
+}
+
+struct TPushTopicOwnerLeaseReq {
+ 1: required list ownerLeases
+}
+
struct TPushConsumerGroupMetaReq {
1: required list consumerGroupMetas
}
@@ -1233,6 +1248,12 @@ service IDataNodeRPCService {
*/
TPushTopicMetaResp pushMultiTopicMeta(TPushMultiTopicMetaReq req)
+ /**
+ * Renew topic owner leases on DataNodes via the dedicated subscription owner heartbeat
+ * (independent from the node heartbeat). Carries relative remaining durations.
+ */
+ common.TSStatus pushTopicOwnerLease(TPushTopicOwnerLeaseReq req)
+
/**
* Send consumerGroupMetas to DataNodes, for synchronization
*/