This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new a321fa34d5d Subscription: preserve topic updates during owner transfer
(#18353)
a321fa34d5d is described below
commit a321fa34d5db3b64d42010c983563e6fc6c2e5e9
Author: Caideyipi <[email protected]>
AuthorDate: Fri Jul 31 10:57:07 2026 +0800
Subscription: preserve topic updates during owner transfer (#18353)
* Subscription: preserve topic updates during owner transfer
* Subscription: serialize concurrent topic alterations
---
.../iotdb/confignode/manager/ProcedureManager.java | 40 ++++--
.../subscription/SubscriptionCoordinator.java | 69 +++++++++-
.../subscription/topic/AlterTopicProcedure.java | 66 ++++++++-
.../procedure/store/ProcedureFactory.java | 7 +-
.../confignode/procedure/store/ProcedureType.java | 1 +
.../subscription/SubscriptionCoordinatorTest.java | 152 +++++++++++++++++++++
.../topic/AlterTopicProcedureTest.java | 82 ++++++++---
7 files changed, 374 insertions(+), 43 deletions(-)
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
index e6a8476500f..bd987cd42a6 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java
@@ -59,6 +59,7 @@ import
org.apache.iotdb.confignode.consensus.request.write.datanode.RemoveDataNo
import
org.apache.iotdb.confignode.consensus.request.write.procedure.UpdateProcedurePlan;
import
org.apache.iotdb.confignode.consensus.request.write.region.CreateRegionGroupsPlan;
import org.apache.iotdb.confignode.i18n.ManagerMessages;
+import
org.apache.iotdb.confignode.manager.subscription.SubscriptionCoordinator;
import org.apache.iotdb.confignode.persistence.ProcedureInfo;
import org.apache.iotdb.confignode.procedure.PartitionTableAutoCleaner;
import org.apache.iotdb.confignode.procedure.Procedure;
@@ -1915,21 +1916,18 @@ public class ProcedureManager {
}
public TSStatus alterTopic(TAlterTopicReq req) {
+ final SubscriptionCoordinator subscriptionCoordinator =
+ configManager.getSubscriptionManager().getSubscriptionCoordinator();
+ subscriptionCoordinator.lockTopicAlteration(req.getTopicName());
boolean isOwnerLeaseRenewalBlocked = false;
try {
isOwnerLeaseRenewalBlocked =
- configManager
- .getSubscriptionManager()
- .getSubscriptionCoordinator()
- .blockOwnerLeaseRenewalIfOwnerTransfer(req);
+ subscriptionCoordinator.blockOwnerLeaseRenewalIfOwnerTransfer(req);
// Owner transfers wait for the previous owner's lease to drain (lease
duration + one
// heartbeat interval, measured on the ConfigNode clock) inside the call
below before the
// updated meta is built; epoch fencing on DataNodes guarantees
correctness in the meantime.
final TopicMeta updatedTopicMeta =
- configManager
- .getSubscriptionManager()
- .getSubscriptionCoordinator()
- .buildAlteredTopicMetaAfterOwnerLeaseExpired(req);
+
subscriptionCoordinator.buildAlteredTopicMetaAfterOwnerLeaseExpired(req);
if (updatedTopicMeta == null) {
return new TSStatus(TSStatusCode.ALTER_TOPIC_ERROR.getStatusCode())
.setMessage(
@@ -1938,9 +1936,27 @@ public class ProcedureManager {
req.getTopicName()));
}
+ final Map<String, String> attributesBeforeInjection =
+ new HashMap<>(updatedTopicMeta.getConfig().getAttribute());
injectTreeViewSourceAttributes(updatedTopicMeta.getConfig().getAttribute());
- AlterTopicProcedure procedure = new
AlterTopicProcedure(updatedTopicMeta);
+ final Map<String, String> updatedTopicAttributes = new HashMap<>();
+ if (Objects.nonNull(req.getTopicAttributes())) {
+ updatedTopicAttributes.putAll(req.getTopicAttributes());
+ }
+ updatedTopicMeta
+ .getConfig()
+ .getAttribute()
+ .forEach(
+ (key, value) -> {
+ if (!attributesBeforeInjection.containsKey(key)
+ || !Objects.equals(attributesBeforeInjection.get(key),
value)) {
+ updatedTopicAttributes.put(key, value);
+ }
+ });
+
+ AlterTopicProcedure procedure =
+ new AlterTopicProcedure(updatedTopicMeta, updatedTopicAttributes);
executor.submitProcedure(procedure);
TSStatus status = waitingProcedureFinished(procedure);
if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
@@ -1953,11 +1969,9 @@ public class ProcedureManager {
.setMessage(e.getMessage());
} finally {
if (isOwnerLeaseRenewalBlocked) {
- configManager
- .getSubscriptionManager()
- .getSubscriptionCoordinator()
- .unblockOwnerLeaseRenewal(req.getTopicName());
+ subscriptionCoordinator.unblockOwnerLeaseRenewal(req.getTopicName());
}
+ subscriptionCoordinator.unlockTopicAlteration(req.getTopicName());
}
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionCoordinator.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionCoordinator.java
index 6fbda89257b..1d988eabc93 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionCoordinator.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionCoordinator.java
@@ -65,6 +65,7 @@ import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantLock;
public class SubscriptionCoordinator {
@@ -80,11 +81,23 @@ public class SubscriptionCoordinator {
private final SubscriptionMetaSyncer subscriptionMetaSyncer;
private final SubscriptionOwnerLeaseSyncer subscriptionOwnerLeaseSyncer;
+
+ // Serialize client ALTER TOPIC requests per topic. Besides preventing
ordinary partial updates
+ // from being built from the same snapshot, this also ensures that only one
owner-transfer
+ // request can own the renewal-block entry for a topic at a time.
+ private final Map<String, TopicAlterationLock> topicAlterationLocks = new
HashMap<>();
+
// topicName -> blockSinceMs (ConfigNode local clock when owner-lease
renewal was stopped for an
// in-flight owner transfer). Used to skip renewal and to bound the
admission wait.
private final Map<String, Long> blockedOwnerLeaseRenewalTopics =
Collections.synchronizedMap(new HashMap<>());
+ private static class TopicAlterationLock {
+
+ private final ReentrantLock lock = new ReentrantLock(true);
+ private int referenceCount;
+ }
+
public SubscriptionCoordinator(ConfigManager configManager, SubscriptionInfo
subscriptionInfo) {
this.configManager = configManager;
this.subscriptionInfo = subscriptionInfo;
@@ -134,6 +147,30 @@ public class SubscriptionCoordinator {
return coordinatorLock.isLocked();
}
+ public void lockTopicAlteration(final String topicName) {
+ final TopicAlterationLock topicAlterationLock;
+ synchronized (topicAlterationLocks) {
+ topicAlterationLock =
+ topicAlterationLocks.computeIfAbsent(topicName, ignored -> new
TopicAlterationLock());
+ topicAlterationLock.referenceCount++;
+ }
+ topicAlterationLock.lock.lock();
+ }
+
+ public void unlockTopicAlteration(final String topicName) {
+ final TopicAlterationLock topicAlterationLock;
+ synchronized (topicAlterationLocks) {
+ topicAlterationLock = topicAlterationLocks.get(topicName);
+ }
+
+ topicAlterationLock.lock.unlock();
+ synchronized (topicAlterationLocks) {
+ if (--topicAlterationLock.referenceCount == 0) {
+ topicAlterationLocks.remove(topicName, topicAlterationLock);
+ }
+ }
+ }
+
/////////////////////////////// Meta sync ///////////////////////////////
public void startSubscriptionMetaSync() {
@@ -185,7 +222,10 @@ public class SubscriptionCoordinator {
public boolean blockOwnerLeaseRenewalIfOwnerTransfer(TAlterTopicReq req) {
final TopicMeta currentTopicMeta =
subscriptionInfo.deepCopyTopicMeta(req.getTopicName());
- final TopicMeta updatedTopicMeta = buildAlteredTopicMeta(req);
+ final TopicMeta updatedTopicMeta =
+ Objects.isNull(currentTopicMeta)
+ ? null
+ :
currentTopicMeta.deepCopyWithUpdatedAttributes(req.getTopicAttributes());
if (Objects.isNull(currentTopicMeta)
|| Objects.isNull(updatedTopicMeta)
|| Objects.equals(currentTopicMeta.getOwnerId(),
updatedTopicMeta.getOwnerId())) {
@@ -211,7 +251,10 @@ public class SubscriptionCoordinator {
public TopicMeta buildAlteredTopicMetaAfterOwnerLeaseExpired(TAlterTopicReq
req)
throws InterruptedException {
final TopicMeta currentTopicMeta =
subscriptionInfo.deepCopyTopicMeta(req.getTopicName());
- final TopicMeta updatedTopicMeta = buildAlteredTopicMeta(req);
+ final TopicMeta updatedTopicMeta =
+ Objects.isNull(currentTopicMeta)
+ ? null
+ :
currentTopicMeta.deepCopyWithUpdatedAttributes(req.getTopicAttributes());
if (Objects.isNull(currentTopicMeta)
|| Objects.isNull(updatedTopicMeta)
|| Objects.equals(currentTopicMeta.getOwnerId(),
updatedTopicMeta.getOwnerId())) {
@@ -220,20 +263,32 @@ public class SubscriptionCoordinator {
}
final Long leaseDurationMs = currentTopicMeta.getOwnerLeaseDurationMs();
- final Long blockSinceMs =
blockedOwnerLeaseRenewalTopics.get(req.getTopicName());
- if (Objects.isNull(leaseDurationMs) || Objects.isNull(blockSinceMs)) {
- // No lease configured (no drain to wait for) or renewal not blocked:
epoch fencing applies on
- // reachable DataNodes; nothing further to wait on here.
+ if (Objects.isNull(leaseDurationMs)) {
+ // No lease configured, so there is nothing to drain.
return updatedTopicMeta;
}
+ waitForOwnerLeaseExpiration(req.getTopicName(), leaseDurationMs);
+
+ // Another alteration may have completed while this owner transfer was
waiting for the old
+ // lease to drain. Rebuild from the latest TopicMeta so that this request
only applies its own
+ // attributes instead of restoring the stale snapshot captured before the
wait.
+ return buildAlteredTopicMeta(req);
+ }
+
+ void waitForOwnerLeaseExpiration(final String topicName, final long
leaseDurationMs)
+ throws InterruptedException {
+ final Long blockSinceMs = blockedOwnerLeaseRenewalTopics.get(topicName);
+ if (Objects.isNull(blockSinceMs)) {
+ return;
+ }
+
final long drainDeadlineMs =
blockSinceMs + leaseDurationMs +
SubscriptionOwnerLeaseSyncer.getHeartbeatIntervalMs();
long remainingMs;
while ((remainingMs = drainDeadlineMs - System.currentTimeMillis()) > 0) {
Thread.sleep(Math.min(remainingMs, 1000L));
}
- return updatedTopicMeta;
}
public TopicMeta buildAlteredTopicMeta(TAlterTopicReq req) {
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedure.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedure.java
index 4a85db0458a..6e9c94dcca4 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedure.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedure.java
@@ -41,7 +41,9 @@ import org.slf4j.LoggerFactory;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
import java.util.concurrent.atomic.AtomicReference;
@@ -53,15 +55,35 @@ public class AlterTopicProcedure extends
AbstractOperateSubscriptionProcedure {
private TopicMeta existedTopicMeta;
+ // Non-null only for client ALTER TOPIC requests. These are merged again
after the procedure has
+ // acquired the subscription lock, so a request cannot overwrite attributes
committed after its
+ // initial TopicMeta snapshot was built.
+ private Map<String, String> updatedTopicAttributes;
+
public AlterTopicProcedure() {
super();
}
+ public AlterTopicProcedure(final boolean shouldMergeUpdatedTopicAttributes) {
+ super();
+ if (shouldMergeUpdatedTopicAttributes) {
+ updatedTopicAttributes = new HashMap<>();
+ }
+ }
+
public AlterTopicProcedure(TopicMeta updatedTopicMeta) {
super();
this.updatedTopicMeta = updatedTopicMeta;
}
+ public AlterTopicProcedure(
+ TopicMeta updatedTopicMeta, Map<String, String> updatedTopicAttributes) {
+ super();
+ this.updatedTopicMeta = updatedTopicMeta;
+ this.updatedTopicAttributes =
+ Objects.isNull(updatedTopicAttributes) ? null : new
HashMap<>(updatedTopicAttributes);
+ }
+
/** This is only used when the subscription info lock is held by another
procedure. */
public AlterTopicProcedure(
TopicMeta updatedTopicMeta, AtomicReference<SubscriptionInfo>
subscriptionInfo) {
@@ -70,6 +92,15 @@ public class AlterTopicProcedure extends
AbstractOperateSubscriptionProcedure {
this.subscriptionInfo = subscriptionInfo;
}
+ /** This is only used when the subscription info lock is held by another
procedure. */
+ public AlterTopicProcedure(
+ TopicMeta updatedTopicMeta,
+ Map<String, String> updatedTopicAttributes,
+ AtomicReference<SubscriptionInfo> subscriptionInfo) {
+ this(updatedTopicMeta, updatedTopicAttributes);
+ this.subscriptionInfo = subscriptionInfo;
+ }
+
/** This should be called after {@link #executeFromValidate}. */
public TopicMeta getExistedTopicMeta() {
return existedTopicMeta;
@@ -80,6 +111,10 @@ public class AlterTopicProcedure extends
AbstractOperateSubscriptionProcedure {
return updatedTopicMeta;
}
+ public boolean shouldMergeUpdatedTopicAttributes() {
+ return Objects.nonNull(updatedTopicAttributes);
+ }
+
@Override
protected SubscriptionOperation getOperation() {
return SubscriptionOperation.ALTER_TOPIC;
@@ -89,9 +124,12 @@ public class AlterTopicProcedure extends
AbstractOperateSubscriptionProcedure {
public boolean executeFromValidate(ConfigNodeProcedureEnv env) throws
SubscriptionException {
LOGGER.info(ProcedureMessages.ALTERTOPICPROCEDURE_EXECUTEFROMVALIDATE);
- subscriptionInfo.get().validateBeforeAlteringTopic(updatedTopicMeta);
+ existedTopicMeta =
subscriptionInfo.get().deepCopyTopicMeta(updatedTopicMeta.getTopicName());
+ if (Objects.nonNull(updatedTopicAttributes) &&
Objects.nonNull(existedTopicMeta)) {
+ updatedTopicMeta =
existedTopicMeta.deepCopyWithUpdatedAttributes(updatedTopicAttributes);
+ }
- existedTopicMeta =
subscriptionInfo.get().getTopicMeta(updatedTopicMeta.getTopicName());
+ subscriptionInfo.get().validateBeforeAlteringTopic(updatedTopicMeta);
return true;
}
@@ -196,7 +234,11 @@ public class AlterTopicProcedure extends
AbstractOperateSubscriptionProcedure {
@Override
public void serialize(DataOutputStream stream) throws IOException {
- stream.writeShort(ProcedureType.ALTER_TOPIC_PROCEDURE.getTypeCode());
+ stream.writeShort(
+ (shouldMergeUpdatedTopicAttributes()
+ ? ProcedureType.ALTER_TOPIC_WITH_ATTRIBUTES_PROCEDURE
+ : ProcedureType.ALTER_TOPIC_PROCEDURE)
+ .getTypeCode());
super.serialize(stream);
ReadWriteIOUtils.write(updatedTopicMeta != null, stream);
@@ -208,6 +250,10 @@ public class AlterTopicProcedure extends
AbstractOperateSubscriptionProcedure {
if (existedTopicMeta != null) {
existedTopicMeta.serialize(stream);
}
+
+ if (shouldMergeUpdatedTopicAttributes()) {
+ ReadWriteIOUtils.write(updatedTopicAttributes, stream);
+ }
}
@Override
@@ -221,6 +267,10 @@ public class AlterTopicProcedure extends
AbstractOperateSubscriptionProcedure {
if (ReadWriteIOUtils.readBool(byteBuffer)) {
existedTopicMeta = TopicMeta.deserialize(byteBuffer);
}
+
+ if (shouldMergeUpdatedTopicAttributes()) {
+ updatedTopicAttributes = ReadWriteIOUtils.readMap(byteBuffer);
+ }
}
@Override
@@ -236,12 +286,18 @@ public class AlterTopicProcedure extends
AbstractOperateSubscriptionProcedure {
&& Objects.equals(getCurrentState(), that.getCurrentState())
&& getCycles() == that.getCycles()
&& Objects.equals(updatedTopicMeta, that.updatedTopicMeta)
- && Objects.equals(existedTopicMeta, that.existedTopicMeta);
+ && Objects.equals(existedTopicMeta, that.existedTopicMeta)
+ && Objects.equals(updatedTopicAttributes, that.updatedTopicAttributes);
}
@Override
public int hashCode() {
return Objects.hash(
- getProcId(), getCurrentState(), getCycles(), updatedTopicMeta,
existedTopicMeta);
+ getProcId(),
+ getCurrentState(),
+ getCycles(),
+ updatedTopicMeta,
+ existedTopicMeta,
+ updatedTopicAttributes);
}
}
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java
index 4cc3af5480b..aadc40f310c 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureFactory.java
@@ -382,6 +382,9 @@ public class ProcedureFactory implements IProcedureFactory {
case ALTER_TOPIC_PROCEDURE:
procedure = new AlterTopicProcedure();
break;
+ case ALTER_TOPIC_WITH_ATTRIBUTES_PROCEDURE:
+ procedure = new AlterTopicProcedure(true);
+ break;
case TOPIC_META_SYNC_PROCEDURE:
procedure = new TopicMetaSyncProcedure();
break;
@@ -546,7 +549,9 @@ public class ProcedureFactory implements IProcedureFactory {
} else if (procedure instanceof DropTopicProcedure) {
return ProcedureType.DROP_TOPIC_PROCEDURE;
} else if (procedure instanceof AlterTopicProcedure) {
- return ProcedureType.ALTER_TOPIC_PROCEDURE;
+ return ((AlterTopicProcedure)
procedure).shouldMergeUpdatedTopicAttributes()
+ ? ProcedureType.ALTER_TOPIC_WITH_ATTRIBUTES_PROCEDURE
+ : ProcedureType.ALTER_TOPIC_PROCEDURE;
} else if (procedure instanceof TopicMetaSyncProcedure) {
return ProcedureType.TOPIC_META_SYNC_PROCEDURE;
} else if (procedure instanceof CreateSubscriptionProcedure) {
diff --git
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureType.java
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureType.java
index c6b885bfac2..c29d1afd672 100644
---
a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureType.java
+++
b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/store/ProcedureType.java
@@ -170,6 +170,7 @@ public enum ProcedureType {
CONSUMER_GROUP_META_SYNC_PROCEDURE((short) 1509),
COMMIT_PROGRESS_SYNC_PROCEDURE((short) 1510),
SUBSCRIPTION_HANDLE_LEADER_CHANGE_PROCEDURE((short) 1511),
+ ALTER_TOPIC_WITH_ATTRIBUTES_PROCEDURE((short) 1512),
/** Other */
@TestOnly
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionCoordinatorTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionCoordinatorTest.java
new file mode 100644
index 00000000000..12b1b3dd2ae
--- /dev/null
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionCoordinatorTest.java
@@ -0,0 +1,152 @@
+/*
+ * 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.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.manager.ConfigManager;
+import org.apache.iotdb.confignode.persistence.subscription.SubscriptionInfo;
+import org.apache.iotdb.confignode.rpc.thrift.TAlterTopicReq;
+import org.apache.iotdb.rpc.TSStatusCode;
+import org.apache.iotdb.rpc.subscription.config.TopicConstant;
+
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+
+public class SubscriptionCoordinatorTest {
+
+ @Test
+ public void testTopicAlterationLockSerializesOnlyTheSameTopic() throws
Exception {
+ final SubscriptionCoordinator coordinator =
+ new SubscriptionCoordinator(Mockito.mock(ConfigManager.class), new
SubscriptionInfo());
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ final CountDownLatch sameTopicAttemptStarted = new CountDownLatch(1);
+ final CountDownLatch sameTopicLockAcquired = new CountDownLatch(1);
+ boolean topic1LockHeld = true;
+
+ coordinator.lockTopicAlteration("topic1");
+ try {
+ final Future<?> sameTopicAlteration =
+ executor.submit(
+ () -> {
+ sameTopicAttemptStarted.countDown();
+ coordinator.lockTopicAlteration("topic1");
+ try {
+ sameTopicLockAcquired.countDown();
+ } finally {
+ coordinator.unlockTopicAlteration("topic1");
+ }
+ });
+
+ Assert.assertTrue(sameTopicAttemptStarted.await(5, TimeUnit.SECONDS));
+ Assert.assertFalse(sameTopicLockAcquired.await(100,
TimeUnit.MILLISECONDS));
+
+ coordinator.lockTopicAlteration("topic2");
+ coordinator.unlockTopicAlteration("topic2");
+
+ coordinator.unlockTopicAlteration("topic1");
+ topic1LockHeld = false;
+ Assert.assertTrue(sameTopicLockAcquired.await(5, TimeUnit.SECONDS));
+ sameTopicAlteration.get(5, TimeUnit.SECONDS);
+ } finally {
+ if (topic1LockHeld) {
+ coordinator.unlockTopicAlteration("topic1");
+ }
+ executor.shutdownNow();
+ }
+ }
+
+ @Test
+ public void testOwnerTransferPreservesConcurrentAlterationDuringLeaseWait()
throws Exception {
+ final String topicName = "test_topic";
+ final String initialColumnFilter = "old-column-filter";
+ final String latestColumnFilter = "latest-column-filter";
+ final long ownerLeaseDurationMs = 60_000L;
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+
+ final Map<String, String> initialAttributes = new HashMap<>();
+ initialAttributes.put(TopicConstant.OWNER_ID_KEY, "owner1");
+ initialAttributes.put(TopicConstant.OWNER_EPOCH_KEY, "1");
+ initialAttributes.put(
+ TopicConstant.OWNER_LEASE_DURATION_MS_KEY,
String.valueOf(ownerLeaseDurationMs));
+ initialAttributes.put(TopicConstant.COLUMN_FILTER_KEY,
initialColumnFilter);
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ subscriptionInfo
+ .createTopic(new CreateTopicPlan(new TopicMeta(topicName, 1L,
initialAttributes)))
+ .getCode());
+
+ final SubscriptionCoordinator coordinator =
+ new SubscriptionCoordinator(Mockito.mock(ConfigManager.class),
subscriptionInfo) {
+ @Override
+ void waitForOwnerLeaseExpiration(
+ final String waitingTopicName, final long
waitingLeaseDurationMs) {
+ Assert.assertEquals(topicName, waitingTopicName);
+ Assert.assertEquals(ownerLeaseDurationMs, waitingLeaseDurationMs);
+
+ final Map<String, String> updatedAttributes = new HashMap<>();
+ updatedAttributes.put(TopicConstant.COLUMN_FILTER_KEY,
latestColumnFilter);
+ final TopicMeta concurrentlyUpdatedTopicMeta =
+ subscriptionInfo.deepCopyTopicMetaWithUpdatedAttributes(
+ topicName, updatedAttributes);
+ Assert.assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ subscriptionInfo
+ .alterTopic(new
AlterTopicPlan(concurrentlyUpdatedTopicMeta))
+ .getCode());
+ }
+ };
+
+ final Map<String, String> ownerTransferAttributes = new HashMap<>();
+ ownerTransferAttributes.put(TopicConstant.OWNER_ID_KEY, "owner2");
+ ownerTransferAttributes.put(TopicConstant.OWNER_EPOCH_KEY, "2");
+ ownerTransferAttributes.put(
+ TopicConstant.OWNER_LEASE_DURATION_MS_KEY,
String.valueOf(ownerLeaseDurationMs));
+ final TAlterTopicReq ownerTransferRequest =
+ new TAlterTopicReq()
+ .setTopicName(topicName)
+ .setTopicAttributes(ownerTransferAttributes)
+ .setSubscribedConsumerGroupIds(Collections.emptySet());
+
+
Assert.assertTrue(coordinator.blockOwnerLeaseRenewalIfOwnerTransfer(ownerTransferRequest));
+ try {
+ final TopicMeta result =
+
coordinator.buildAlteredTopicMetaAfterOwnerLeaseExpired(ownerTransferRequest);
+
+ Assert.assertEquals("owner2", result.getOwnerId());
+ Assert.assertEquals(2L, result.getOwnerEpoch());
+ Assert.assertEquals(
+ latestColumnFilter,
result.getConfig().getString(TopicConstant.COLUMN_FILTER_KEY));
+ } finally {
+ coordinator.unblockOwnerLeaseRenewal(topicName);
+ }
+ }
+}
diff --git
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedureTest.java
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedureTest.java
index b943e4edd97..3f553cb6013 100644
---
a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedureTest.java
+++
b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/AlterTopicProcedureTest.java
@@ -20,7 +20,11 @@
package org.apache.iotdb.confignode.procedure.impl.subscription.topic;
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.persistence.subscription.SubscriptionInfo;
import org.apache.iotdb.confignode.procedure.store.ProcedureFactory;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.tsfile.utils.PublicBAOS;
import org.junit.Test;
@@ -29,32 +33,76 @@ import java.io.DataOutputStream;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
public class AlterTopicProcedureTest {
+
@Test
- public void serializeDeserializeTest() {
- PublicBAOS byteArrayOutputStream = new PublicBAOS();
- DataOutputStream outputStream = new
DataOutputStream(byteArrayOutputStream);
+ public void serializeDeserializeTest() throws Exception {
+ final Map<String, String> topicAttributes = new HashMap<>();
+ topicAttributes.put("path", "root.db1.**");
+ assertSerializeDeserialize(
+ new AlterTopicProcedure(new TopicMeta("test_topic", 1,
topicAttributes)));
+ }
- Map<String, String> topicAttributes = new HashMap<>();
+ @Test
+ public void serializeDeserializeWithUpdatedAttributesTest() throws Exception
{
+ final Map<String, String> topicAttributes = new HashMap<>();
topicAttributes.put("path", "root.db1.**");
+ final Map<String, String> updatedTopicAttributes = new HashMap<>();
+ updatedTopicAttributes.put("processor", "processor1");
+ assertSerializeDeserialize(
+ new AlterTopicProcedure(
+ new TopicMeta("test_topic", 1, topicAttributes),
updatedTopicAttributes));
+ }
+
+ private void assertSerializeDeserialize(final AlterTopicProcedure procedure)
throws Exception {
+ final PublicBAOS byteArrayOutputStream = new PublicBAOS();
+ final DataOutputStream outputStream = new
DataOutputStream(byteArrayOutputStream);
+ procedure.serialize(outputStream);
+ final ByteBuffer buffer =
+ ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0,
byteArrayOutputStream.size());
+ final AlterTopicProcedure deserializedProcedure =
+ (AlterTopicProcedure) ProcedureFactory.getInstance().create(buffer);
+ assertEquals(procedure, deserializedProcedure);
+ }
+
+ @Test
+ public void testRebaseUpdatedAttributesDuringValidate() throws Exception {
+ final String topicName = "test_topic";
+ final SubscriptionInfo subscriptionInfo = new SubscriptionInfo();
+ final Map<String, String> initialAttributes = new HashMap<>();
+ initialAttributes.put("path", "root.db1.**");
+ assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ subscriptionInfo
+ .createTopic(new CreateTopicPlan(new TopicMeta(topicName, 1,
initialAttributes)))
+ .getCode());
+
+ final Map<String, String> requestAttributes = new HashMap<>();
+ requestAttributes.put("processor", "processor1");
+ final TopicMeta staleUpdatedTopicMeta =
+ subscriptionInfo.deepCopyTopicMetaWithUpdatedAttributes(topicName,
requestAttributes);
- AlterTopicProcedure proc =
- new AlterTopicProcedure(new TopicMeta("test_topic", 1,
topicAttributes));
+ final Map<String, String> concurrentAttributes = new HashMap<>();
+ concurrentAttributes.put("source", "source1");
+ assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ subscriptionInfo
+ .alterTopic(
+ new AlterTopicPlan(
+ subscriptionInfo.deepCopyTopicMetaWithUpdatedAttributes(
+ topicName, concurrentAttributes)))
+ .getCode());
- try {
- proc.serialize(outputStream);
- ByteBuffer buffer =
- ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0,
byteArrayOutputStream.size());
- AlterTopicProcedure proc2 =
- (AlterTopicProcedure) ProcedureFactory.getInstance().create(buffer);
+ final AlterTopicProcedure procedure =
+ new AlterTopicProcedure(
+ staleUpdatedTopicMeta, requestAttributes, new
AtomicReference<>(subscriptionInfo));
+ procedure.executeFromValidate(null);
- assertEquals(proc, proc2);
- } catch (Exception e) {
- fail();
- }
+ assertEquals("processor1",
procedure.getUpdatedTopicMeta().getConfig().getString("processor"));
+ assertEquals("source1",
procedure.getUpdatedTopicMeta().getConfig().getString("source"));
}
}
