From c4d0d9df239d7fa1d83599a307327c0a630a2796 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 29 Jul 2026 17:33:33 +0800 Subject: [PATCH 1/2] Subscription: preserve topic updates during owner transfer --- .../subscription/SubscriptionCoordinator.java | 22 +++- .../SubscriptionCoordinatorTest.java | 106 ++++++++++++++++++ 2 files changed, 123 insertions(+), 5 deletions(-) create mode 100644 iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionCoordinatorTest.java 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 6fbda89257b46..c3d3380be11b6 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 @@ -220,20 +220,32 @@ public TopicMeta buildAlteredTopicMetaAfterOwnerLeaseExpired(TAlterTopicReq req) } 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/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 0000000000000..dd7f9682801e9 --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/subscription/SubscriptionCoordinatorTest.java @@ -0,0 +1,106 @@ +/* + * 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; + +public class SubscriptionCoordinatorTest { + + @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 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 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 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); + } + } +} From 22c743be85aee7af24136c26fd873ef568f4f7b2 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 30 Jul 2026 16:15:39 +0800 Subject: [PATCH 2/2] Subscription: serialize concurrent topic alterations --- .../confignode/manager/ProcedureManager.java | 40 ++++++--- .../subscription/SubscriptionCoordinator.java | 47 ++++++++++- .../topic/AlterTopicProcedure.java | 66 +++++++++++++-- .../procedure/store/ProcedureFactory.java | 7 +- .../procedure/store/ProcedureType.java | 1 + .../SubscriptionCoordinatorTest.java | 46 +++++++++++ .../topic/AlterTopicProcedureTest.java | 82 +++++++++++++++---- 7 files changed, 251 insertions(+), 38 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 e6a8476500fd1..bd987cd42a647 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.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 TSStatus createTopic(TCreateTopicReq req) { } 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 TSStatus alterTopic(TAlterTopicReq req) { req.getTopicName())); } + final Map attributesBeforeInjection = + new HashMap<>(updatedTopicMeta.getConfig().getAttribute()); injectTreeViewSourceAttributes(updatedTopicMeta.getConfig().getAttribute()); - AlterTopicProcedure procedure = new AlterTopicProcedure(updatedTopicMeta); + final Map 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 TSStatus alterTopic(TAlterTopicReq req) { .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 c3d3380be11b6..1d988eabc9394 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.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 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 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 boolean isLocked() { 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 TSStatus alterTopic(TAlterTopicReq req) { 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 void unblockOwnerLeaseRenewal(String topicName) { 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())) { 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 4a85db0458a9e..6e9c94dcca420 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 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 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 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) { @@ -70,6 +92,15 @@ public AlterTopicProcedure( this.subscriptionInfo = subscriptionInfo; } + /** This is only used when the subscription info lock is held by another procedure. */ + public AlterTopicProcedure( + TopicMeta updatedTopicMeta, + Map updatedTopicAttributes, + AtomicReference subscriptionInfo) { + this(updatedTopicMeta, updatedTopicAttributes); + this.subscriptionInfo = subscriptionInfo; + } + /** This should be called after {@link #executeFromValidate}. */ public TopicMeta getExistedTopicMeta() { return existedTopicMeta; @@ -80,6 +111,10 @@ public TopicMeta getUpdatedTopicMeta() { return updatedTopicMeta; } + public boolean shouldMergeUpdatedTopicAttributes() { + return Objects.nonNull(updatedTopicAttributes); + } + @Override protected SubscriptionOperation getOperation() { return SubscriptionOperation.ALTER_TOPIC; @@ -89,9 +124,12 @@ protected SubscriptionOperation getOperation() { 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 void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) @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 void serialize(DataOutputStream stream) throws IOException { if (existedTopicMeta != null) { existedTopicMeta.serialize(stream); } + + if (shouldMergeUpdatedTopicAttributes()) { + ReadWriteIOUtils.write(updatedTopicAttributes, stream); + } } @Override @@ -221,6 +267,10 @@ public void deserialize(ByteBuffer byteBuffer) { if (ReadWriteIOUtils.readBool(byteBuffer)) { existedTopicMeta = TopicMeta.deserialize(byteBuffer); } + + if (shouldMergeUpdatedTopicAttributes()) { + updatedTopicAttributes = ReadWriteIOUtils.readMap(byteBuffer); + } } @Override @@ -236,12 +286,18 @@ public boolean equals(Object o) { && 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 4cc3af5480b12..aadc40f310c7b 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 Procedure create(ByteBuffer buffer) throws IOException { 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 static ProcedureType getProcedureType(final Procedure procedure) { } 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 c6b885bfac2b3..c29d1afd67287 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 index dd7f9682801e9..12b1b3dd2aeaa 100644 --- 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 @@ -35,9 +35,55 @@ 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"; 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 b943e4edd973b..3f553cb60135b 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.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 topicAttributes = new HashMap<>(); + topicAttributes.put("path", "root.db1.**"); + assertSerializeDeserialize( + new AlterTopicProcedure(new TopicMeta("test_topic", 1, topicAttributes))); + } - Map topicAttributes = new HashMap<>(); + @Test + public void serializeDeserializeWithUpdatedAttributesTest() throws Exception { + final Map topicAttributes = new HashMap<>(); topicAttributes.put("path", "root.db1.**"); + final Map 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 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 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 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")); } }