This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch feature/initial-retain-progress-after-unsubscribe in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit f3d9020835b0145d013ba0e76967389e40064ea7 Author: Caideyipi <[email protected]> AuthorDate: Mon Oct 26 17:58:54 2026 +0800 Support progress retention for initial subscription topics --- .../it/local/IoTDBSubscriptionBasicIT.java | 134 +++++++++++ .../iotdb/confignode/i18n/ConfigNodeMessages.java | 4 + .../iotdb/confignode/i18n/ConfigNodeMessages.java | 4 + .../persistence/subscription/SubscriptionInfo.java | 103 ++++++-- .../consumer/AlterConsumerGroupProcedure.java | 3 + .../subscription/CreateSubscriptionProcedure.java | 25 +- .../subscription/DropSubscriptionProcedure.java | 5 + .../subscription/topic/AlterTopicProcedure.java | 240 +++++++++++++++--- .../subscription/topic/DropTopicProcedure.java | 267 +++++++++++++++++++-- .../procedure/store/ProcedureFactory.java | 21 +- .../confignode/procedure/store/ProcedureType.java | 3 + .../SubscriptionInfoTopicValidationTest.java | 37 ++- .../consumer/DropConsumerProcedureTest.java | 61 +++++ .../CreateSubscriptionProcedureTest.java | 47 ++++ .../DropSubscriptionProcedureTest.java | 34 +++ .../topic/AlterTopicProcedureTest.java | 7 + .../subscription/topic/DropTopicProcedureTest.java | 244 +++++++++++++++++++ .../agent/SubscriptionBrokerAgent.java | 8 + .../agent/SubscriptionConsumerAgent.java | 55 ++++- .../db/subscription/broker/SubscriptionBroker.java | 17 +- .../agent/SubscriptionConsumerAgentTest.java | 89 +++++++ .../meta/consumer/CommitProgressKeeper.java | 14 ++ .../meta/consumer/ConsumerGroupMeta.java | 53 +++- .../consumer/ConsumerGroupDeSerTest.java | 35 +++ 24 files changed, 1415 insertions(+), 95 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionBasicIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionBasicIT.java index bed9721f3aa..18e0fb296be 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionBasicIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/IoTDBSubscriptionBasicIT.java @@ -54,6 +54,8 @@ import org.slf4j.LoggerFactory; import java.io.IOException; import java.time.Duration; +import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Properties; @@ -63,6 +65,7 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.locks.LockSupport; import java.util.stream.Collectors; +import java.util.stream.LongStream; import static org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant.AWAIT; import static org.junit.Assert.fail; @@ -430,6 +433,137 @@ public class IoTDBSubscriptionBasicIT extends AbstractSubscriptionLocalIT { } } + @Test + public void testInitialTopicRetainsProgressAfterUnsubscribe() throws Exception { + final String topicName = "topic_initial_retained_progress"; + final String host = EnvFactory.getEnv().getIP(); + final int port = Integer.parseInt(EnvFactory.getEnv().getPort()); + try (final ISession session = EnvFactory.getEnv().getSessionConnection()) { + session.executeNonQueryStatement("create database root.retain_progress"); + for (int i = 0; i < 20; ++i) { + session.executeNonQueryStatement( + String.format("insert into root.retain_progress.d1(time, s1) values (%s, 1)", i)); + } + session.executeNonQueryStatement("flush"); + } + + try (final SubscriptionTreeSession session = new SubscriptionTreeSession(host, port)) { + session.open(); + final Properties properties = new Properties(); + properties.put(TopicConstant.MODE_KEY, TopicConstant.MODE_INITIAL_VALUE); + properties.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY, true); + session.createTopic(topicName, properties); + } + + try (final SubscriptionTreePullConsumer consumer = + new SubscriptionTreePullConsumer.Builder() + .host(host) + .port(port) + .consumerId("consumer_retained_progress") + .consumerGroupId("group_retained_progress") + .autoCommit(false) + .buildPullConsumer()) { + consumer.open(); + consumer.subscribe(topicName); + + Assert.assertEquals( + LongStream.range(0, 20).boxed().collect(Collectors.toList()), + pollTimestamps(consumer, 20, true)); + + try (final ISession session = EnvFactory.getEnv().getSessionConnection()) { + session.executeNonQueryStatement( + "insert into root.retain_progress.d1(time, s1) values (20, 1)"); + session.executeNonQueryStatement("flush"); + } + + // Leave this delivery uncommitted. The server must requeue it when the last consumer leaves. + Assert.assertEquals(Collections.singletonList(20L), pollTimestamps(consumer, 1, false)); + + consumer.unsubscribe(topicName); + try (final ISession session = EnvFactory.getEnv().getSessionConnection()) { + session.executeNonQueryStatement( + "insert into root.retain_progress.d1(time, s1) values (21, 1)"); + session.executeNonQueryStatement("flush"); + } + + consumer.subscribe(topicName); + Assert.assertEquals(Arrays.asList(20L, 21L), pollTimestamps(consumer, 2, true)); + consumer.unsubscribe(topicName); + } + } + + @Test + public void testInitialTopicDoesNotRetainProgressByDefault() throws Exception { + final String topicName = "topic_initial_unretained_progress"; + final String host = EnvFactory.getEnv().getIP(); + final int port = Integer.parseInt(EnvFactory.getEnv().getPort()); + try (final ISession session = EnvFactory.getEnv().getSessionConnection()) { + session.executeNonQueryStatement("create database root.unretained_progress"); + for (int i = 0; i < 3; ++i) { + session.executeNonQueryStatement( + String.format("insert into root.unretained_progress.d1(time, s1) values (%s, 1)", i)); + } + session.executeNonQueryStatement("flush"); + } + + try (final SubscriptionTreeSession session = new SubscriptionTreeSession(host, port)) { + session.open(); + final Properties properties = new Properties(); + properties.put(TopicConstant.MODE_KEY, TopicConstant.MODE_INITIAL_VALUE); + session.createTopic(topicName, properties); + } + + try (final SubscriptionTreePullConsumer consumer = + new SubscriptionTreePullConsumer.Builder() + .host(host) + .port(port) + .consumerId("consumer_unretained_progress") + .consumerGroupId("group_unretained_progress") + .autoCommit(false) + .buildPullConsumer()) { + consumer.open(); + consumer.subscribe(topicName); + Assert.assertEquals(Arrays.asList(0L, 1L, 2L), pollTimestamps(consumer, 3, true)); + consumer.unsubscribe(topicName); + + try (final ISession session = EnvFactory.getEnv().getSessionConnection()) { + session.executeNonQueryStatement( + "insert into root.unretained_progress.d1(time, s1) values (3, 1)"); + session.executeNonQueryStatement("flush"); + } + + consumer.subscribe(topicName); + Assert.assertEquals(Arrays.asList(0L, 1L, 2L, 3L), pollTimestamps(consumer, 4, true)); + consumer.unsubscribe(topicName); + } + } + + private static List<Long> pollTimestamps( + final SubscriptionTreePullConsumer consumer, final int expectedRows, final boolean commit) { + final List<Long> timestamps = new ArrayList<>(); + AWAIT.untilAsserted( + () -> { + final List<SubscriptionMessage> messages = consumer.poll(Duration.ofSeconds(1)); + for (final SubscriptionMessage message : messages) { + for (final ResultSet dataSet : message.getResultSets()) { + try (final ResultSet result = dataSet) { + final SubscriptionRecordHandler.SubscriptionResultSet records = + (SubscriptionRecordHandler.SubscriptionResultSet) result; + while (records.hasNext()) { + timestamps.add(records.nextRecord().getTimestamp()); + } + } + } + } + if (commit) { + consumer.commitSync(messages); + } + Assert.assertEquals(expectedRows, timestamps.size()); + }); + Collections.sort(timestamps); + return timestamps; + } + @Ignore @Test public void testTsFileDeduplication() { diff --git a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java index cfde56c6419..72d6493104f 100644 --- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java +++ b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java @@ -20,6 +20,10 @@ package org.apache.iotdb.confignode.i18n; public final class ConfigNodeMessages { + public static final String + EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ARG_IS_NOT_SUPPORTED_FOR_SNAPSHOT_TOPICS_894FDD40 = + "Failed to create or alter topic, %s is not supported for snapshot topics"; + public static final String EXCEPTION_CONFIGNODE_SIMPLECONSENSUS_WRITES_ARE_DISABLED_AFTER_A_WAL_CLEANUP_FAILURE_RESTART_IS_REQUIRED_D52F0107 = "ConfigNode SimpleConsensus writes are disabled after a WAL cleanup failure; restart is required."; public static final String EXCEPTION_FAILED_TO_CLOSE_THE_CONFIGNODE_SIMPLECONSENSUS_LOG_WRITER_BC0F417C = diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java index ddf8be67cef..865fd6ac795 100644 --- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java +++ b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ConfigNodeMessages.java @@ -20,6 +20,10 @@ package org.apache.iotdb.confignode.i18n; public final class ConfigNodeMessages { + public static final String + EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ARG_IS_NOT_SUPPORTED_FOR_SNAPSHOT_TOPICS_894FDD40 = + "创建或修改主题失败，快照模式不支持 %s"; + public static final String EXCEPTION_CONFIGNODE_SIMPLECONSENSUS_WRITES_ARE_DISABLED_AFTER_A_WAL_CLEANUP_FAILURE_RESTART_IS_REQUIRED_D52F0107 = "ConfigNode SimpleConsensus 因 WAL 清理失败已禁用写入，需要重启。"; public static final String EXCEPTION_FAILED_TO_CLOSE_THE_CONFIGNODE_SIMPLECONSENSUS_LOG_WRITER_BC0F417C = 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 79a735ded7f..2ce4781a7e6 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 @@ -411,6 +411,7 @@ public class SubscriptionInfo implements SnapshotProcessor { validateColumnFilter(topicConfig); validateTagFilter(topicConfig); validateIncrementalTopicRetentionConfig(topicConfig); + validateRetainProgressAfterUnsubscribe(topicConfig); validateRecycleUncommittedEventInterval(topicConfig); final Long ownerLeaseDurationMs = @@ -654,9 +655,7 @@ public class SubscriptionInfo implements SnapshotProcessor { private void validateIncrementalTopicRetentionConfig(final TopicConfig topicConfig) throws SubscriptionException { if (!hasTopicAttributeIgnoreCase(topicConfig, TopicConstant.RETENTION_BYTES_KEY) - && !hasTopicAttributeIgnoreCase(topicConfig, TopicConstant.RETENTION_MS_KEY) - && !hasTopicAttributeIgnoreCase( - topicConfig, TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY)) { + && !hasTopicAttributeIgnoreCase(topicConfig, TopicConstant.RETENTION_MS_KEY)) { return; } @@ -665,17 +664,13 @@ public class SubscriptionInfo implements SnapshotProcessor { String.format( ConfigNodeMessages .EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ATTRIBUTES_ARG_ARE_ONLY_SUPPORTED_FOR_INCREMENTAL_TOPICS_55CB0BF2, - List.of( - TopicConstant.RETENTION_BYTES_KEY, - TopicConstant.RETENTION_MS_KEY, - TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY)); + List.of(TopicConstant.RETENTION_BYTES_KEY, TopicConstant.RETENTION_MS_KEY)); LOGGER.warn(exceptionMessage); throw new SubscriptionException(exceptionMessage); } validateRetentionValue(topicConfig, TopicConstant.RETENTION_BYTES_KEY); validateRetentionValue(topicConfig, TopicConstant.RETENTION_MS_KEY); - validateRetainProgressAfterUnsubscribe(topicConfig); } private void validateRetainProgressAfterUnsubscribe(final TopicConfig topicConfig) @@ -684,6 +679,15 @@ public class SubscriptionInfo implements SnapshotProcessor { topicConfig, TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY)) { return; } + if (topicConfig.isSnapshotMode()) { + final String exceptionMessage = + String.format( + ConfigNodeMessages + .EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ARG_IS_NOT_SUPPORTED_FOR_SNAPSHOT_TOPICS_894FDD40, + TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY); + LOGGER.warn(exceptionMessage); + throw new SubscriptionException(exceptionMessage); + } final String value = getTopicAttributeIgnoreCase( topicConfig, TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY); @@ -958,12 +962,21 @@ public class SubscriptionInfo implements SnapshotProcessor { if (Objects.nonNull(oldTopicMeta) && oldTopicMeta.getConfig().isProgressRetainedAfterUnsubscribe() - && !plan.getTopicMeta().getConfig().isProgressRetainedAfterUnsubscribe()) { + && (!plan.getTopicMeta().getConfig().isProgressRetainedAfterUnsubscribe() + || (oldTopicMeta.getConfig().isInitialMode() + && !plan.getTopicMeta().getConfig().isInitialMode()))) { for (final ConsumerGroupMeta consumerGroupMeta : consumerGroupMetaKeeper.getAllConsumerGroupMeta()) { if (consumerGroupMeta.visibleUnder(isTableModel) && !consumerGroupMeta.isTopicSubscribedByConsumerGroup( plan.getTopicMeta().getTopicName())) { + if (oldTopicMeta.getConfig().isInitialMode() + && consumerGroupMeta + .getRetainedInitialTopicNames() + .contains(plan.getTopicMeta().getTopicName())) { + // The AlterTopicProcedure drops the retained Pipe before clearing this marker. + continue; + } commitProgressKeeper.removeTopicProgress( consumerGroupMeta.getConsumerGroupId(), plan.getTopicMeta().getTopicName()); removeEmptyConsumerGroupWithoutProgress(consumerGroupMeta); @@ -1004,25 +1017,52 @@ public class SubscriptionInfo implements SnapshotProcessor { public TSStatus dropTopic(DropTopicPlan plan) { acquireWriteLock(); try { + final TopicMeta topicMeta = + plan.isTableModelSet() + ? topicMetaKeeper.getTopicMeta(plan.getTopicName(), plan.isTableModel()) + : topicMetaKeeper.getTopicMeta(plan.getTopicName()); + final boolean isTableModel = + plan.isTableModelSet() + ? plan.isTableModel() + : Objects.nonNull(topicMeta) && topicMeta.visibleUnderTableModel(); + final boolean isModelScoped = plan.isTableModelSet() || Objects.nonNull(topicMeta); for (final ConsumerGroupMeta consumerGroupMeta : consumerGroupMetaKeeper.getAllConsumerGroupMeta()) { - if (!plan.isTableModelSet() || consumerGroupMeta.visibleUnder(plan.isTableModel())) { + if (!isModelScoped || consumerGroupMeta.visibleUnder(isTableModel)) { + if (isModelScoped + && consumerGroupMeta.getRetainedInitialTopicNames().contains(plan.getTopicName())) { + // The DropTopicProcedure drops the retained Pipe before clearing this marker. + continue; + } commitProgressKeeper.removeTopicProgress( consumerGroupMeta.getConsumerGroupId(), plan.getTopicName()); removeEmptyConsumerGroupWithoutProgress(consumerGroupMeta); } } - if (plan.isTableModelSet()) { - topicMetaKeeper.removeTopicMeta(plan.getTopicName(), plan.isTableModel()); - } else { - topicMetaKeeper.removeTopicMeta(plan.getTopicName()); - } + topicMetaKeeper.removeTopicMeta(plan.getTopicName(), isTableModel); return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode()); } finally { releaseWriteLock(); } } + public List<ConsumerGroupMeta> getConsumerGroupsRetainingInitialTopic( + final String topicName, final boolean isTableModel) { + acquireReadLock(); + try { + final List<ConsumerGroupMeta> consumerGroupMetas = new ArrayList<>(); + for (final ConsumerGroupMeta meta : consumerGroupMetaKeeper.getAllConsumerGroupMeta()) { + if (meta.visibleUnder(isTableModel) + && meta.getRetainedInitialTopicNames().contains(topicName)) { + consumerGroupMetas.add(meta.deepCopy()); + } + } + return consumerGroupMetas; + } finally { + releaseReadLock(); + } + } + public TSStatus handleTopicMetaChanges(TopicHandleMetaChangePlan plan) { acquireWriteLock(); try { @@ -1185,6 +1225,12 @@ public class SubscriptionInfo implements SnapshotProcessor { ? consumerGroupMetaKeeper.getConsumerGroupMeta(consumerGroupId) : null; if (Objects.nonNull(currentConsumerGroupMeta)) { + final Set<String> removedRetainedTopics = + new HashSet<>(currentConsumerGroupMeta.getRetainedInitialTopicNames()); + removedRetainedTopics.removeAll(consumerGroupMeta.getRetainedInitialTopicNames()); + removedRetainedTopics.removeAll(consumerGroupMeta.getSubscribedTopicNames()); + removedRetainedTopics.forEach( + topicName -> commitProgressKeeper.removeTopicProgress(consumerGroupId, topicName)); ConsumerGroupMeta.getTopicsUnsubByGroup(currentConsumerGroupMeta, consumerGroupMeta) .forEach( topicName -> { @@ -1196,7 +1242,7 @@ public class SubscriptionInfo implements SnapshotProcessor { commitProgressKeeper.removeTopicProgress(consumerGroupId, topicName); } }); - if (consumerGroupMeta.isEmpty()) { + if (consumerGroupMeta.isEmpty() && !currentConsumerGroupMeta.isEmpty()) { final Set<String> retainedTopicNames = new HashSet<>(); for (final TopicMeta topicMeta : topicMetaKeeper.getAllTopicMeta()) { if (topicMeta.visibleUnderTableModel() == currentConsumerGroupMeta.visibleUnder(true) @@ -1210,6 +1256,7 @@ public class SubscriptionInfo implements SnapshotProcessor { } consumerGroupMetaKeeper.removeConsumerGroupMeta(consumerGroupId); if (!consumerGroupMeta.isEmpty() + || !consumerGroupMeta.getRetainedInitialTopicNames().isEmpty() || commitProgressKeeper.hasConsumerGroupProgress(consumerGroupId)) { consumerGroupMetaKeeper.addConsumerGroupMeta(consumerGroupId, consumerGroupMeta); } @@ -1222,11 +1269,35 @@ public class SubscriptionInfo implements SnapshotProcessor { private void removeEmptyConsumerGroupWithoutProgress(final ConsumerGroupMeta consumerGroupMeta) { if (consumerGroupMeta.isEmpty() + && consumerGroupMeta.getRetainedInitialTopicNames().isEmpty() && !commitProgressKeeper.hasConsumerGroupProgress(consumerGroupMeta.getConsumerGroupId())) { consumerGroupMetaKeeper.removeConsumerGroupMeta(consumerGroupMeta.getConsumerGroupId()); } } + public void retainInitialSubscriptionProgress( + final ConsumerGroupMeta existingMeta, final ConsumerGroupMeta updatedMeta) { + if (Objects.isNull(existingMeta)) { + return; + } + acquireReadLock(); + try { + for (final String topicName : + ConsumerGroupMeta.getTopicsUnsubByGroup(existingMeta, updatedMeta)) { + final TopicMeta topicMeta = + topicMetaKeeper.getTopicMeta(topicName, existingMeta.visibleUnder(true)); + if (Objects.nonNull(topicMeta) + && topicMeta.getConfig().isInitialMode() + && topicMeta.getConfig().isProgressRetainedAfterUnsubscribe()) { + updatedMeta.retainInitialTopic(topicName); + } + } + updatedMeta.getSubscribedTopicNames().forEach(updatedMeta::removeRetainedInitialTopic); + } finally { + releaseReadLock(); + } + } + public TSStatus handleConsumerGroupMetaChanges(ConsumerGroupHandleMetaChangePlan plan) { acquireWriteLock(); try { diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/consumer/AlterConsumerGroupProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/consumer/AlterConsumerGroupProcedure.java index faf9cce8a68..0f6e5924e57 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/consumer/AlterConsumerGroupProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/consumer/AlterConsumerGroupProcedure.java @@ -81,6 +81,9 @@ public class AlterConsumerGroupProcedure extends AbstractOperateSubscriptionProc LOGGER.info(ProcedureMessages.ALTERCONSUMERGROUPPROCEDURE_EXECUTEFROMVALIDATE_TRY_TO_VALIDATE); validateAndGetOldAndNewMeta(env); + subscriptionInfo + .get() + .retainInitialSubscriptionProgress(existingConsumerGroupMeta, updatedConsumerGroupMeta); return true; } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/CreateSubscriptionProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/CreateSubscriptionProcedure.java index 04e29edbb94..f8a19838144 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/CreateSubscriptionProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/CreateSubscriptionProcedure.java @@ -135,10 +135,15 @@ public class CreateSubscriptionProcedure extends AbstractOperateSubscriptionAndP final String pipeName = PipeStaticMeta.generateSubscriptionPipeName(topicName, consumerGroupId); - if (!subscriptionInfo.get().isTopicSubscribedByConsumerGroup(topicName, consumerGroupId) - // even if there existed subscription meta, if there is no corresponding pipe meta, it - // will try to create the pipe - || !pipeTaskInfo.get().isPipeExisted(pipeName, topicMeta.visibleUnderTableModel())) { + final boolean pipeExists = + pipeTaskInfo.get().isPipeExisted(pipeName, topicMeta.visibleUnderTableModel()); + final boolean topicAlreadySubscribed = + consumerGroupMeta.getSubscribedTopicNames().contains(topicName); + final boolean topicProgressRetained = + consumerGroupMeta.getRetainedInitialTopicNames().contains(topicName); + // A retained initial subscription keeps the original Pipe and its sink queue alive. Reuse it + // only while the group is subscribed or has an explicit retained-progress marker. + if (!pipeExists || (!topicAlreadySubscribed && !topicProgressRetained)) { createPipeProcedures.add( new CreatePipeProcedureV2( new TCreatePipeReq() @@ -227,12 +232,12 @@ public class CreateSubscriptionProcedure extends AbstractOperateSubscriptionAndP consensusTopicNames); } - // Push pipe meta to data nodes (only for non-consensus pipe-based topics) - if (!createPipeProcedures.isEmpty()) { - final List<PipeStaticMeta> pipeStaticMetas = - createPipeProcedures.stream() - .map(CreatePipeProcedureV2::getPipeStaticMeta) - .collect(Collectors.toList()); + // Push pipe meta to data nodes (only for newly created non-consensus pipes). + final List<PipeStaticMeta> pipeStaticMetas = + createPipeProcedures.stream() + .map(CreatePipeProcedureV2::getPipeStaticMeta) + .collect(Collectors.toList()); + if (!pipeStaticMetas.isEmpty()) { final String exceptionMessage = AbstractOperatePipeProcedureV2.parsePushPipeMetaExceptionForPipe( null, pushMultiPipeMetaToDataNodes(pipeStaticMetas, env)); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java index b4c16985aef..ad6004302ea 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedure.java @@ -121,6 +121,11 @@ public class DropSubscriptionProcedure extends AbstractOperateSubscriptionAndPip continue; } + if (topicMeta.getConfig().isInitialMode() + && topicMeta.getConfig().isProgressRetainedAfterUnsubscribe()) { + continue; + } + // Topic will be subscribed by no consumers in this group dropPipeProcedures.add( new DropPipeProcedureV2( 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 fb7ff2f6332..f034e6b9e05 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 @@ -20,14 +20,26 @@ package org.apache.iotdb.confignode.procedure.impl.subscription.topic; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta; +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.topic.TopicMeta; +import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan; +import org.apache.iotdb.confignode.consensus.request.write.pipe.task.CreatePipePlanV2; +import org.apache.iotdb.confignode.consensus.request.write.pipe.task.DropPipePlanV2; +import org.apache.iotdb.confignode.consensus.request.write.pipe.task.OperateMultiplePipesPlanV2; +import org.apache.iotdb.confignode.consensus.request.write.subscription.consumer.AlterConsumerGroupPlan; +import org.apache.iotdb.confignode.consensus.request.write.subscription.consumer.runtime.CommitProgressHandleMetaChangePlan; import org.apache.iotdb.confignode.consensus.request.write.subscription.topic.AlterTopicPlan; import org.apache.iotdb.confignode.i18n.ConfigNodeMessages; import org.apache.iotdb.confignode.i18n.ProcedureMessages; import org.apache.iotdb.confignode.persistence.subscription.SubscriptionInfo; import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; -import org.apache.iotdb.confignode.procedure.impl.subscription.AbstractOperateSubscriptionProcedure; +import org.apache.iotdb.confignode.procedure.impl.pipe.AbstractOperatePipeProcedureV2; +import org.apache.iotdb.confignode.procedure.impl.pipe.task.DropPipeProcedureV2; import org.apache.iotdb.confignode.procedure.impl.subscription.SubscriptionOperation; +import org.apache.iotdb.confignode.procedure.impl.subscription.subscription.AbstractOperateSubscriptionAndPipeProcedure; import org.apache.iotdb.confignode.procedure.store.ProcedureType; import org.apache.iotdb.consensus.exception.ConsensusException; import org.apache.iotdb.rpc.RpcUtils; @@ -41,13 +53,14 @@ import org.slf4j.LoggerFactory; import java.io.DataOutputStream; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.ArrayList; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Objects; import java.util.concurrent.atomic.AtomicReference; -public class AlterTopicProcedure extends AbstractOperateSubscriptionProcedure { +public class AlterTopicProcedure extends AbstractOperateSubscriptionAndPipeProcedure { private static final Logger LOGGER = LoggerFactory.getLogger(AlterTopicProcedure.class); @@ -60,6 +73,11 @@ public class AlterTopicProcedure extends AbstractOperateSubscriptionProcedure { // initial TopicMeta snapshot was built. private Map<String, String> updatedTopicAttributes; + private final List<ConsumerGroupMeta> retainedConsumerGroupMetas = new ArrayList<>(); + private final List<DropPipeProcedureV2> retainedPipeProcedures = new ArrayList<>(); + private Map<String, ByteBuffer> retainedProgress = new HashMap<>(); + private boolean serializeRetainedProgressState; + public AlterTopicProcedure() { super(); } @@ -71,6 +89,13 @@ public class AlterTopicProcedure extends AbstractOperateSubscriptionProcedure { } } + public AlterTopicProcedure( + final boolean shouldMergeUpdatedTopicAttributes, + final boolean serializeRetainedProgressState) { + this(shouldMergeUpdatedTopicAttributes); + this.serializeRetainedProgressState = serializeRetainedProgressState; + } + public AlterTopicProcedure(TopicMeta updatedTopicMeta) { super(); this.updatedTopicMeta = updatedTopicMeta; @@ -115,6 +140,12 @@ public class AlterTopicProcedure extends AbstractOperateSubscriptionProcedure { return Objects.nonNull(updatedTopicAttributes); } + public boolean shouldSerializeRetainedProgressState() { + return serializeRetainedProgressState + || !retainedConsumerGroupMetas.isEmpty() + || !retainedPipeProcedures.isEmpty(); + } + @Override protected SubscriptionOperation getOperation() { return SubscriptionOperation.ALTER_TOPIC; @@ -124,6 +155,10 @@ public class AlterTopicProcedure extends AbstractOperateSubscriptionProcedure { public boolean executeFromValidate(ConfigNodeProcedureEnv env) throws SubscriptionException { LOGGER.info(ProcedureMessages.ALTERTOPICPROCEDURE_EXECUTEFROMVALIDATE); + retainedConsumerGroupMetas.clear(); + retainedPipeProcedures.clear(); + retainedProgress.clear(); + existedTopicMeta = subscriptionInfo .get() @@ -136,23 +171,75 @@ public class AlterTopicProcedure extends AbstractOperateSubscriptionProcedure { subscriptionInfo.get().validateBeforeAlteringTopic(updatedTopicMeta); + if (Objects.nonNull(existedTopicMeta) + && existedTopicMeta.getConfig().isInitialMode() + && existedTopicMeta.getConfig().isProgressRetainedAfterUnsubscribe() + && !updatedTopicMeta.getConfig().isProgressRetainedAfterUnsubscribe()) { + prepareRetainedProgressCleanup(env); + } + return true; } + private void prepareRetainedProgressCleanup(final ConfigNodeProcedureEnv env) { + final String topicName = updatedTopicMeta.getTopicName(); + final boolean isTableModel = updatedTopicMeta.visibleUnderTableModel(); + for (final ConsumerGroupMeta consumerGroupMeta : + subscriptionInfo.get().getConsumerGroupsRetainingInitialTopic(topicName, isTableModel)) { + retainedConsumerGroupMetas.add(consumerGroupMeta); + retainedProgress.putAll( + subscriptionInfo + .get() + .getCommitProgressKeeper() + .getTopicProgress(consumerGroupMeta.getConsumerGroupId(), topicName)); + final String pipeName = + PipeStaticMeta.generateSubscriptionPipeName( + topicName, consumerGroupMeta.getConsumerGroupId()); + if (pipeTaskInfo.get().isPipeExisted(pipeName, isTableModel)) { + final DropPipeProcedureV2 dropPipeProcedure = + new DropPipeProcedureV2(pipeName, isTableModel, pipeTaskInfo); + dropPipeProcedure.executeFromValidateTask(env); + dropPipeProcedure.executeFromCalculateInfoForTask(env); + retainedPipeProcedures.add(dropPipeProcedure); + } + } + } + @Override public void executeFromOperateOnConfigNodes(ConfigNodeProcedureEnv env) throws SubscriptionException { LOGGER.info( ProcedureMessages.ALTERTOPICPROCEDURE_EXECUTEFROMOPERATEONCONFIGNODES_TRY_TO_ALTER_TOPIC); - TSStatus response; + if (!retainedPipeProcedures.isEmpty()) { + final List<ConfigPhysicalPlan> dropPipePlans = + retainedPipeProcedures.stream() + .map( + procedure -> + new DropPipePlanV2( + procedure.getPipeName(), updatedTopicMeta.visibleUnderTableModel())) + .collect(java.util.stream.Collectors.toList()); + writeConsensusPlan(env, new OperateMultiplePipesPlanV2(dropPipePlans)); + } + for (final ConsumerGroupMeta consumerGroupMeta : getUpdatedRetainedConsumerGroupMetas()) { + writeConsensusPlan(env, new AlterConsumerGroupPlan(consumerGroupMeta)); + } + writeAlterTopicPlan(env, updatedTopicMeta); + } + + private void writeAlterTopicPlan(final ConfigNodeProcedureEnv env, final TopicMeta topicMeta) + throws SubscriptionException { + writeConsensusPlan(env, new AlterTopicPlan(topicMeta)); + } + + private void writeConsensusPlan(final ConfigNodeProcedureEnv env, final ConfigPhysicalPlan plan) + throws SubscriptionException { + final TSStatus response; try { - response = - env.getConfigManager().getConsensusManager().write(new AlterTopicPlan(updatedTopicMeta)); + response = env.getConfigManager().getConsensusManager().write(plan); } catch (ConsensusException e) { LOGGER.warn(ConfigNodeMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE, e); - response = - new TSStatus(TSStatusCode.ALTER_TOPIC_ERROR.getStatusCode()).setMessage(e.getMessage()); + throw new SubscriptionException(e.getMessage()); } if (response.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { throw new SubscriptionException( @@ -164,6 +251,16 @@ public class AlterTopicProcedure extends AbstractOperateSubscriptionProcedure { } } + private List<ConsumerGroupMeta> getUpdatedRetainedConsumerGroupMetas() { + final List<ConsumerGroupMeta> updatedMetas = new ArrayList<>(); + for (final ConsumerGroupMeta existingMeta : retainedConsumerGroupMetas) { + final ConsumerGroupMeta updatedMeta = existingMeta.deepCopy(); + updatedMeta.removeRetainedInitialTopic(updatedTopicMeta.getTopicName()); + updatedMetas.add(updatedMeta); + } + return updatedMetas; + } + @Override public void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) throws SubscriptionException, IOException { @@ -171,6 +268,24 @@ public class AlterTopicProcedure extends AbstractOperateSubscriptionProcedure { ProcedureMessages.ALTERTOPICPROCEDURE_EXECUTEFROMOPERATEONDATANODES, updatedTopicMeta.getTopicName()); + if (!retainedPipeProcedures.isEmpty()) { + final String exceptionMessage = + AbstractOperatePipeProcedureV2.parsePushPipeMetaExceptionForPipe( + null, dropMultiPipeOnDataNodes(retainedPipeProcedures, env)); + if (!exceptionMessage.isEmpty()) { + throw new SubscriptionException(exceptionMessage); + } + } + + for (final ConsumerGroupMeta consumerGroupMeta : getUpdatedRetainedConsumerGroupMetas()) { + final List<TSStatus> consumerGroupStatuses = + env.pushSingleConsumerGroupOnDataNode(consumerGroupMeta.serialize()); + if (RpcUtils.squashResponseStatusList(consumerGroupStatuses).getCode() + != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + throw new SubscriptionException(consumerGroupStatuses.toString()); + } + } + final List<TSStatus> statuses = env.pushSingleTopicOnDataNode(updatedTopicMeta.serialize()); if (RpcUtils.squashResponseStatusList(statuses).getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { @@ -198,23 +313,35 @@ public class AlterTopicProcedure extends AbstractOperateSubscriptionProcedure { ProcedureMessages.ALTERTOPICPROCEDURE_ROLLBACKFROMOPERATEONCONFIGNODES, updatedTopicMeta.getTopicName()); - TSStatus response; - try { - response = - env.getConfigManager().getConsensusManager().write(new AlterTopicPlan(existedTopicMeta)); - } catch (ConsensusException e) { - LOGGER.warn(ConfigNodeMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE, e); - response = - new TSStatus(TSStatusCode.ALTER_TOPIC_ERROR.getStatusCode()).setMessage(e.getMessage()); + writeAlterTopicPlan(env, existedTopicMeta); + if (!retainedPipeProcedures.isEmpty()) { + for (final DropPipeProcedureV2 dropPipeProcedure : retainedPipeProcedures) { + final PipeMeta pipeMeta = dropPipeProcedure.getPipeMetaToDrop(); + if (!pipeTaskInfo + .get() + .isPipeExisted( + pipeMeta.getStaticMeta().getPipeName(), + pipeMeta.getStaticMeta().visibleUnderTableModel())) { + writeConsensusPlan( + env, new CreatePipePlanV2(pipeMeta.getStaticMeta(), pipeMeta.getRuntimeMeta())); + } + } } - if (response.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - throw new SubscriptionException( - String.format( - ProcedureMessages.FAILED_TO_ROLLBACK_FROM_ALTERING_TOPIC_ON_CONFIG_NODES_BECAUSE, - updatedTopicMeta, - existedTopicMeta, - response)); + for (final ConsumerGroupMeta consumerGroupMeta : retainedConsumerGroupMetas) { + writeConsensusPlan(env, new AlterConsumerGroupPlan(consumerGroupMeta)); } + restoreRetainedProgress(env); + } + + private void restoreRetainedProgress(final ConfigNodeProcedureEnv env) + throws SubscriptionException { + if (retainedProgress.isEmpty()) { + return; + } + final Map<String, ByteBuffer> progress = + subscriptionInfo.get().getCommitProgressKeeper().getAllRegionProgress(); + progress.putAll(retainedProgress); + writeConsensusPlan(env, new CommitProgressHandleMetaChangePlan(progress)); } @Override @@ -224,6 +351,15 @@ public class AlterTopicProcedure extends AbstractOperateSubscriptionProcedure { ProcedureMessages.ALTERTOPICPROCEDURE_ROLLBACKFROMOPERATEONDATANODES, updatedTopicMeta.getTopicName()); + for (final ConsumerGroupMeta consumerGroupMeta : retainedConsumerGroupMetas) { + final List<TSStatus> consumerGroupStatuses = + env.pushSingleConsumerGroupOnDataNode(consumerGroupMeta.serialize()); + if (RpcUtils.squashResponseStatusList(consumerGroupStatuses).getCode() + != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + throw new SubscriptionException(consumerGroupStatuses.toString()); + } + } + final List<TSStatus> statuses = env.pushSingleTopicOnDataNode(existedTopicMeta.serialize()); if (RpcUtils.squashResponseStatusList(statuses).getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { @@ -235,15 +371,19 @@ public class AlterTopicProcedure extends AbstractOperateSubscriptionProcedure { existedTopicMeta, statuses)); } + if (!retainedPipeProcedures.isEmpty()) { + final String exceptionMessage = + AbstractOperatePipeProcedureV2.parsePushPipeMetaExceptionForPipe( + null, AbstractOperatePipeProcedureV2.pushPipeMetaToDataNodes(env, pipeTaskInfo)); + if (!exceptionMessage.isEmpty()) { + throw new SubscriptionException(exceptionMessage); + } + } } @Override public void serialize(DataOutputStream stream) throws IOException { - stream.writeShort( - (shouldMergeUpdatedTopicAttributes() - ? ProcedureType.ALTER_TOPIC_WITH_ATTRIBUTES_PROCEDURE - : ProcedureType.ALTER_TOPIC_PROCEDURE) - .getTypeCode()); + stream.writeShort(getProcedureType().getTypeCode()); super.serialize(stream); ReadWriteIOUtils.write(updatedTopicMeta != null, stream); @@ -259,6 +399,28 @@ public class AlterTopicProcedure extends AbstractOperateSubscriptionProcedure { if (shouldMergeUpdatedTopicAttributes()) { ReadWriteIOUtils.write(updatedTopicAttributes, stream); } + if (shouldSerializeRetainedProgressState()) { + ReadWriteIOUtils.write(retainedConsumerGroupMetas.size(), stream); + for (final ConsumerGroupMeta consumerGroupMeta : retainedConsumerGroupMetas) { + consumerGroupMeta.serialize(stream); + } + ReadWriteIOUtils.write(retainedPipeProcedures.size(), stream); + for (final DropPipeProcedureV2 pipeProcedure : retainedPipeProcedures) { + pipeProcedure.serialize(stream); + } + CommitProgressKeeper.serializeRegionProgressMapToStream(retainedProgress, stream); + } + } + + private ProcedureType getProcedureType() { + if (shouldSerializeRetainedProgressState()) { + return shouldMergeUpdatedTopicAttributes() + ? ProcedureType.ALTER_TOPIC_WITH_ATTRIBUTES_AND_RETAINED_PROGRESS_PROCEDURE + : ProcedureType.ALTER_TOPIC_WITH_RETAINED_PROGRESS_PROCEDURE; + } + return shouldMergeUpdatedTopicAttributes() + ? ProcedureType.ALTER_TOPIC_WITH_ATTRIBUTES_PROCEDURE + : ProcedureType.ALTER_TOPIC_PROCEDURE; } @Override @@ -276,6 +438,20 @@ public class AlterTopicProcedure extends AbstractOperateSubscriptionProcedure { if (shouldMergeUpdatedTopicAttributes()) { updatedTopicAttributes = ReadWriteIOUtils.readMap(byteBuffer); } + if (shouldSerializeRetainedProgressState()) { + int size = ReadWriteIOUtils.readInt(byteBuffer); + for (int i = 0; i < size; i++) { + retainedConsumerGroupMetas.add(ConsumerGroupMeta.deserialize(byteBuffer)); + } + size = ReadWriteIOUtils.readInt(byteBuffer); + for (int i = 0; i < size; i++) { + ReadWriteIOUtils.readShort(byteBuffer); + final DropPipeProcedureV2 pipeProcedure = new DropPipeProcedureV2(); + pipeProcedure.deserialize(byteBuffer); + retainedPipeProcedures.add(pipeProcedure); + } + retainedProgress = CommitProgressKeeper.deserializeRegionProgressFromBuffer(byteBuffer); + } } @Override @@ -292,7 +468,10 @@ public class AlterTopicProcedure extends AbstractOperateSubscriptionProcedure { && getCycles() == that.getCycles() && Objects.equals(updatedTopicMeta, that.updatedTopicMeta) && Objects.equals(existedTopicMeta, that.existedTopicMeta) - && Objects.equals(updatedTopicAttributes, that.updatedTopicAttributes); + && Objects.equals(updatedTopicAttributes, that.updatedTopicAttributes) + && Objects.equals(retainedConsumerGroupMetas, that.retainedConsumerGroupMetas) + && Objects.equals(retainedPipeProcedures, that.retainedPipeProcedures) + && Objects.equals(retainedProgress, that.retainedProgress); } @Override @@ -303,6 +482,9 @@ public class AlterTopicProcedure extends AbstractOperateSubscriptionProcedure { getCycles(), updatedTopicMeta, existedTopicMeta, - updatedTopicAttributes); + updatedTopicAttributes, + retainedConsumerGroupMetas, + retainedPipeProcedures, + retainedProgress); } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/DropTopicProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/DropTopicProcedure.java index 8a95811fce5..bee5210b1ce 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/DropTopicProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/DropTopicProcedure.java @@ -20,12 +20,26 @@ package org.apache.iotdb.confignode.procedure.impl.subscription.topic; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta; +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.topic.TopicMeta; +import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan; +import org.apache.iotdb.confignode.consensus.request.write.pipe.task.CreatePipePlanV2; +import org.apache.iotdb.confignode.consensus.request.write.pipe.task.DropPipePlanV2; +import org.apache.iotdb.confignode.consensus.request.write.pipe.task.OperateMultiplePipesPlanV2; +import org.apache.iotdb.confignode.consensus.request.write.subscription.consumer.AlterConsumerGroupPlan; +import org.apache.iotdb.confignode.consensus.request.write.subscription.consumer.runtime.CommitProgressHandleMetaChangePlan; +import org.apache.iotdb.confignode.consensus.request.write.subscription.topic.CreateTopicPlan; import org.apache.iotdb.confignode.consensus.request.write.subscription.topic.DropTopicPlan; import org.apache.iotdb.confignode.i18n.ConfigNodeMessages; import org.apache.iotdb.confignode.i18n.ProcedureMessages; import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; -import org.apache.iotdb.confignode.procedure.impl.subscription.AbstractOperateSubscriptionProcedure; +import org.apache.iotdb.confignode.procedure.impl.pipe.AbstractOperatePipeProcedureV2; +import org.apache.iotdb.confignode.procedure.impl.pipe.task.DropPipeProcedureV2; import org.apache.iotdb.confignode.procedure.impl.subscription.SubscriptionOperation; +import org.apache.iotdb.confignode.procedure.impl.subscription.subscription.AbstractOperateSubscriptionAndPipeProcedure; import org.apache.iotdb.confignode.procedure.store.ProcedureType; import org.apache.iotdb.consensus.exception.ConsensusException; import org.apache.iotdb.rpc.RpcUtils; @@ -39,16 +53,24 @@ import org.slf4j.LoggerFactory; import java.io.DataOutputStream; import java.io.IOException; import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Objects; -public class DropTopicProcedure extends AbstractOperateSubscriptionProcedure { +public class DropTopicProcedure extends AbstractOperateSubscriptionAndPipeProcedure { private static final Logger LOGGER = LoggerFactory.getLogger(DropTopicProcedure.class); private String topicName; private boolean isTableModel; private boolean isTableModelSet; + private TopicMeta existedTopicMeta; + private final List<ConsumerGroupMeta> retainedConsumerGroupMetas = new ArrayList<>(); + private final List<DropPipeProcedureV2> retainedPipeProcedures = new ArrayList<>(); + private Map<String, ByteBuffer> retainedProgress = new HashMap<>(); + private boolean serializeRetainedProgressState; public DropTopicProcedure() { super(); @@ -65,6 +87,17 @@ public class DropTopicProcedure extends AbstractOperateSubscriptionProcedure { this.isTableModelSet = true; } + public DropTopicProcedure(final boolean serializeRetainedProgressState) { + super(); + this.serializeRetainedProgressState = serializeRetainedProgressState; + } + + public boolean shouldSerializeRetainedProgressState() { + return serializeRetainedProgressState + || !retainedConsumerGroupMetas.isEmpty() + || !retainedPipeProcedures.isEmpty(); + } + @Override protected SubscriptionOperation getOperation() { return SubscriptionOperation.DROP_TOPIC; @@ -74,11 +107,44 @@ public class DropTopicProcedure extends AbstractOperateSubscriptionProcedure { protected boolean executeFromValidate(ConfigNodeProcedureEnv env) throws SubscriptionException { LOGGER.info(ProcedureMessages.DROPTOPICPROCEDURE_EXECUTEFROMVALIDATE, topicName); + retainedConsumerGroupMetas.clear(); + retainedPipeProcedures.clear(); + retainedProgress.clear(); + if (isTableModelSet) { subscriptionInfo.get().validateBeforeDroppingTopic(topicName, isTableModel); } else { subscriptionInfo.get().validateBeforeDroppingTopic(topicName); } + + existedTopicMeta = + isTableModelSet + ? subscriptionInfo.get().deepCopyTopicMeta(topicName, isTableModel) + : subscriptionInfo.get().deepCopyTopicMeta(topicName); + if (Objects.nonNull(existedTopicMeta) + && existedTopicMeta.getConfig().isInitialMode() + && existedTopicMeta.getConfig().isProgressRetainedAfterUnsubscribe()) { + final boolean model = existedTopicMeta.visibleUnderTableModel(); + for (final ConsumerGroupMeta consumerGroupMeta : + subscriptionInfo.get().getConsumerGroupsRetainingInitialTopic(topicName, model)) { + retainedConsumerGroupMetas.add(consumerGroupMeta); + retainedProgress.putAll( + subscriptionInfo + .get() + .getCommitProgressKeeper() + .getTopicProgress(consumerGroupMeta.getConsumerGroupId(), topicName)); + final String pipeName = + PipeStaticMeta.generateSubscriptionPipeName( + topicName, consumerGroupMeta.getConsumerGroupId()); + if (pipeTaskInfo.get().isPipeExisted(pipeName, model)) { + final DropPipeProcedureV2 dropPipeProcedure = + new DropPipeProcedureV2(pipeName, model, pipeTaskInfo); + dropPipeProcedure.executeFromValidateTask(env); + dropPipeProcedure.executeFromCalculateInfoForTask(env); + retainedPipeProcedures.add(dropPipeProcedure); + } + } + } return true; } @@ -87,15 +153,29 @@ public class DropTopicProcedure extends AbstractOperateSubscriptionProcedure { throws SubscriptionException { LOGGER.info(ProcedureMessages.DROPTOPICPROCEDURE_EXECUTEFROMOPERATEONCONFIGNODES, topicName); + final boolean model = getTopicModel(); + writeConsensusPlan( + env, + existedTopicMeta == null && !isTableModelSet + ? new DropTopicPlan(topicName) + : new DropTopicPlan(topicName, model)); + if (!retainedPipeProcedures.isEmpty()) { + final List<ConfigPhysicalPlan> dropPipePlans = + retainedPipeProcedures.stream() + .map(proc -> new DropPipePlanV2(proc.getPipeName(), model)) + .collect(java.util.stream.Collectors.toList()); + writeConsensusPlan(env, new OperateMultiplePipesPlanV2(dropPipePlans)); + } + for (final ConsumerGroupMeta consumerGroupMeta : getUpdatedRetainedConsumerGroupMetas()) { + writeConsensusPlan(env, new AlterConsumerGroupPlan(consumerGroupMeta)); + } + } + + private void writeConsensusPlan(final ConfigNodeProcedureEnv env, final ConfigPhysicalPlan plan) + throws SubscriptionException { TSStatus response; try { - response = - env.getConfigManager() - .getConsensusManager() - .write( - isTableModelSet - ? new DropTopicPlan(topicName, isTableModel) - : new DropTopicPlan(topicName)); + response = env.getConfigManager().getConsensusManager().write(plan); } catch (ConsensusException e) { LOGGER.warn(ConfigNodeMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE, e); response = @@ -108,14 +188,46 @@ public class DropTopicProcedure extends AbstractOperateSubscriptionProcedure { } } + private List<ConsumerGroupMeta> getUpdatedRetainedConsumerGroupMetas() { + final List<ConsumerGroupMeta> updatedMetas = new ArrayList<>(); + for (final ConsumerGroupMeta existingMeta : retainedConsumerGroupMetas) { + final ConsumerGroupMeta updatedMeta = existingMeta.deepCopy(); + updatedMeta.removeRetainedInitialTopic(topicName); + updatedMetas.add(updatedMeta); + } + return updatedMetas; + } + + private boolean getTopicModel() { + return existedTopicMeta == null ? isTableModel : existedTopicMeta.visibleUnderTableModel(); + } + @Override - protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) { + protected void executeFromOperateOnDataNodes(ConfigNodeProcedureEnv env) + throws SubscriptionException, IOException { LOGGER.info(ProcedureMessages.DROPTOPICPROCEDURE_EXECUTEFROMOPERATEONDATANODES, topicName); + if (!retainedPipeProcedures.isEmpty()) { + final String exceptionMessage = + AbstractOperatePipeProcedureV2.parsePushPipeMetaExceptionForPipe( + null, dropMultiPipeOnDataNodes(retainedPipeProcedures, env)); + if (!exceptionMessage.isEmpty()) { + throw new SubscriptionException(exceptionMessage); + } + } + for (final ConsumerGroupMeta consumerGroupMeta : getUpdatedRetainedConsumerGroupMetas()) { + final List<TSStatus> consumerGroupStatuses = + env.pushSingleConsumerGroupOnDataNode(consumerGroupMeta.serialize()); + if (RpcUtils.squashResponseStatusList(consumerGroupStatuses).getCode() + != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + throw new SubscriptionException(consumerGroupStatuses.toString()); + } + } + final List<TSStatus> statuses = - isTableModelSet - ? env.dropSingleTopicOnDataNode(topicName, isTableModel) - : env.dropSingleTopicOnDataNode(topicName); + existedTopicMeta == null && !isTableModelSet + ? env.dropSingleTopicOnDataNode(topicName) + : env.dropSingleTopicOnDataNode(topicName, getTopicModel()); if (RpcUtils.squashResponseStatusList(statuses).getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { // throw exception instead of logging warn, do not rely on metadata synchronization @@ -131,22 +243,98 @@ public class DropTopicProcedure extends AbstractOperateSubscriptionProcedure { } @Override - protected void rollbackFromOperateOnConfigNodes(ConfigNodeProcedureEnv env) { + protected void rollbackFromOperateOnConfigNodes(ConfigNodeProcedureEnv env) + throws SubscriptionException { LOGGER.info(ProcedureMessages.DROPTOPICPROCEDURE_ROLLBACKFROMCREATEONCONFIGNODES, topicName); + if (Objects.nonNull(existedTopicMeta)) { + writeConsensusPlan(env, new CreateTopicPlan(existedTopicMeta)); + for (final DropPipeProcedureV2 dropPipeProcedure : retainedPipeProcedures) { + final PipeMeta pipeMeta = dropPipeProcedure.getPipeMetaToDrop(); + if (!pipeTaskInfo + .get() + .isPipeExisted( + pipeMeta.getStaticMeta().getPipeName(), + pipeMeta.getStaticMeta().visibleUnderTableModel())) { + writeConsensusPlan( + env, new CreatePipePlanV2(pipeMeta.getStaticMeta(), pipeMeta.getRuntimeMeta())); + } + } + for (final ConsumerGroupMeta consumerGroupMeta : retainedConsumerGroupMetas) { + writeConsensusPlan(env, new AlterConsumerGroupPlan(consumerGroupMeta)); + } + restoreRetainedProgress(env); + } + } + + private void restoreRetainedProgress(final ConfigNodeProcedureEnv env) + throws SubscriptionException { + if (retainedProgress.isEmpty()) { + return; + } + final Map<String, ByteBuffer> progress = + subscriptionInfo.get().getCommitProgressKeeper().getAllRegionProgress(); + progress.putAll(retainedProgress); + writeConsensusPlan(env, new CommitProgressHandleMetaChangePlan(progress)); } @Override - protected void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) { + protected void rollbackFromOperateOnDataNodes(ConfigNodeProcedureEnv env) + throws SubscriptionException, IOException { LOGGER.info(ProcedureMessages.DROPTOPICPROCEDURE_ROLLBACKFROMCREATEONDATANODES, topicName); + if (Objects.nonNull(existedTopicMeta)) { + for (final ConsumerGroupMeta consumerGroupMeta : retainedConsumerGroupMetas) { + final List<TSStatus> statuses = + env.pushSingleConsumerGroupOnDataNode(consumerGroupMeta.serialize()); + if (RpcUtils.squashResponseStatusList(statuses).getCode() + != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + throw new SubscriptionException(statuses.toString()); + } + } + final List<TSStatus> statuses = env.pushSingleTopicOnDataNode(existedTopicMeta.serialize()); + if (RpcUtils.squashResponseStatusList(statuses).getCode() + != TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + throw new SubscriptionException(statuses.toString()); + } + if (!retainedPipeProcedures.isEmpty()) { + final String exceptionMessage = + AbstractOperatePipeProcedureV2.parsePushPipeMetaExceptionForPipe( + null, AbstractOperatePipeProcedureV2.pushPipeMetaToDataNodes(env, pipeTaskInfo)); + if (!exceptionMessage.isEmpty()) { + throw new SubscriptionException(exceptionMessage); + } + } + } } @Override public void serialize(DataOutputStream stream) throws IOException { - stream.writeShort(ProcedureType.DROP_TOPIC_PROCEDURE.getTypeCode()); + stream.writeShort( + (shouldSerializeRetainedProgressState() + ? ProcedureType.DROP_TOPIC_WITH_RETAINED_PROGRESS_PROCEDURE + : ProcedureType.DROP_TOPIC_PROCEDURE) + .getTypeCode()); super.serialize(stream); ReadWriteIOUtils.write(topicName, stream); - if (isTableModelSet) { + if (shouldSerializeRetainedProgressState()) { + ReadWriteIOUtils.write(Objects.nonNull(existedTopicMeta), stream); + if (Objects.nonNull(existedTopicMeta)) { + existedTopicMeta.serialize(stream); + } + ReadWriteIOUtils.write(isTableModelSet, stream); + if (isTableModelSet) { + ReadWriteIOUtils.write(isTableModel, stream); + } + ReadWriteIOUtils.write(retainedConsumerGroupMetas.size(), stream); + for (final ConsumerGroupMeta consumerGroupMeta : retainedConsumerGroupMetas) { + consumerGroupMeta.serialize(stream); + } + ReadWriteIOUtils.write(retainedPipeProcedures.size(), stream); + for (final DropPipeProcedureV2 pipeProcedure : retainedPipeProcedures) { + pipeProcedure.serialize(stream); + } + CommitProgressKeeper.serializeRegionProgressMapToStream(retainedProgress, stream); + } else if (isTableModelSet) { ReadWriteIOUtils.write(isTableModel, stream); } } @@ -156,9 +344,31 @@ public class DropTopicProcedure extends AbstractOperateSubscriptionProcedure { super.deserialize(byteBuffer); topicName = ReadWriteIOUtils.readString(byteBuffer); - isTableModelSet = byteBuffer.hasRemaining(); - if (isTableModelSet) { - isTableModel = ReadWriteIOUtils.readBool(byteBuffer); + if (shouldSerializeRetainedProgressState()) { + if (ReadWriteIOUtils.readBool(byteBuffer)) { + existedTopicMeta = TopicMeta.deserialize(byteBuffer); + } + isTableModelSet = ReadWriteIOUtils.readBool(byteBuffer); + if (isTableModelSet) { + isTableModel = ReadWriteIOUtils.readBool(byteBuffer); + } + int size = ReadWriteIOUtils.readInt(byteBuffer); + for (int i = 0; i < size; i++) { + retainedConsumerGroupMetas.add(ConsumerGroupMeta.deserialize(byteBuffer)); + } + size = ReadWriteIOUtils.readInt(byteBuffer); + for (int i = 0; i < size; i++) { + ReadWriteIOUtils.readShort(byteBuffer); + final DropPipeProcedureV2 pipeProcedure = new DropPipeProcedureV2(); + pipeProcedure.deserialize(byteBuffer); + retainedPipeProcedures.add(pipeProcedure); + } + retainedProgress = CommitProgressKeeper.deserializeRegionProgressFromBuffer(byteBuffer); + } else { + isTableModelSet = byteBuffer.hasRemaining(); + if (isTableModelSet) { + isTableModel = ReadWriteIOUtils.readBool(byteBuffer); + } } } @@ -176,12 +386,25 @@ public class DropTopicProcedure extends AbstractOperateSubscriptionProcedure { && getCycles() == that.getCycles() && Objects.equals(topicName, that.topicName) && isTableModel == that.isTableModel - && isTableModelSet == that.isTableModelSet; + && isTableModelSet == that.isTableModelSet + && Objects.equals(existedTopicMeta, that.existedTopicMeta) + && Objects.equals(retainedConsumerGroupMetas, that.retainedConsumerGroupMetas) + && Objects.equals(retainedPipeProcedures, that.retainedPipeProcedures) + && Objects.equals(retainedProgress, that.retainedProgress); } @Override public int hashCode() { return Objects.hash( - getProcId(), getCurrentState(), getCycles(), topicName, isTableModel, isTableModelSet); + getProcId(), + getCurrentState(), + getCycles(), + topicName, + isTableModel, + isTableModelSet, + existedTopicMeta, + retainedConsumerGroupMetas, + retainedPipeProcedures, + retainedProgress); } } 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 0f6b2fa75e1..26d6ae02408 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,12 +382,21 @@ public class ProcedureFactory implements IProcedureFactory { case DROP_TOPIC_PROCEDURE: procedure = new DropTopicProcedure(); break; + case DROP_TOPIC_WITH_RETAINED_PROGRESS_PROCEDURE: + procedure = new DropTopicProcedure(true); + break; case ALTER_TOPIC_PROCEDURE: procedure = new AlterTopicProcedure(); break; case ALTER_TOPIC_WITH_ATTRIBUTES_PROCEDURE: procedure = new AlterTopicProcedure(true); break; + case ALTER_TOPIC_WITH_RETAINED_PROGRESS_PROCEDURE: + procedure = new AlterTopicProcedure(false, true); + break; + case ALTER_TOPIC_WITH_ATTRIBUTES_AND_RETAINED_PROGRESS_PROCEDURE: + procedure = new AlterTopicProcedure(true, true); + break; case TOPIC_META_SYNC_PROCEDURE: procedure = new TopicMetaSyncProcedure(); break; @@ -550,9 +559,17 @@ public class ProcedureFactory implements IProcedureFactory { } else if (procedure instanceof CreateTopicProcedure) { return ProcedureType.CREATE_TOPIC_PROCEDURE; } else if (procedure instanceof DropTopicProcedure) { - return ProcedureType.DROP_TOPIC_PROCEDURE; + return ((DropTopicProcedure) procedure).shouldSerializeRetainedProgressState() + ? ProcedureType.DROP_TOPIC_WITH_RETAINED_PROGRESS_PROCEDURE + : ProcedureType.DROP_TOPIC_PROCEDURE; } else if (procedure instanceof AlterTopicProcedure) { - return ((AlterTopicProcedure) procedure).shouldMergeUpdatedTopicAttributes() + final AlterTopicProcedure alterTopicProcedure = (AlterTopicProcedure) procedure; + if (alterTopicProcedure.shouldSerializeRetainedProgressState()) { + return alterTopicProcedure.shouldMergeUpdatedTopicAttributes() + ? ProcedureType.ALTER_TOPIC_WITH_ATTRIBUTES_AND_RETAINED_PROGRESS_PROCEDURE + : ProcedureType.ALTER_TOPIC_WITH_RETAINED_PROGRESS_PROCEDURE; + } + return alterTopicProcedure.shouldMergeUpdatedTopicAttributes() ? ProcedureType.ALTER_TOPIC_WITH_ATTRIBUTES_PROCEDURE : ProcedureType.ALTER_TOPIC_PROCEDURE; } else if (procedure instanceof TopicMetaSyncProcedure) { 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 7514d06bce4..3656cddc714 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 @@ -172,6 +172,9 @@ public enum ProcedureType { COMMIT_PROGRESS_SYNC_PROCEDURE((short) 1510), SUBSCRIPTION_HANDLE_LEADER_CHANGE_PROCEDURE((short) 1511), ALTER_TOPIC_WITH_ATTRIBUTES_PROCEDURE((short) 1512), + ALTER_TOPIC_WITH_RETAINED_PROGRESS_PROCEDURE((short) 1513), + ALTER_TOPIC_WITH_ATTRIBUTES_AND_RETAINED_PROGRESS_PROCEDURE((short) 1514), + DROP_TOPIC_WITH_RETAINED_PROGRESS_PROCEDURE((short) 1515), /** Other */ @TestOnly 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 fd97bebe74f..caa586b35fd 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 @@ -596,21 +596,48 @@ public class SubscriptionInfoTopicValidationTest { } @Test - public void testRejectRetainingProgressOnInitialTopic() { + public void testAcceptRetainingProgressOnInitialTopic() throws Exception { + final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); + for (final String value : new String[] {"true", " FALSE "}) { + final Map<String, String> attributes = newInitialTableTopicAttributes(); + attributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY, value); + + Assert.assertTrue( + subscriptionInfo.validateBeforeCreatingTopic( + new TCreateTopicReq("initial_topic").setTopicAttributes(attributes))); + } + } + + @Test + public void testRejectRetainingProgressOnSnapshotTopic() { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); final Map<String, String> attributes = newInitialTableTopicAttributes(); + attributes.put(TopicConstant.MODE_KEY, TopicConstant.MODE_SNAPSHOT_VALUE); attributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY, "true"); + assertCreateRejected(subscriptionInfo, attributes, "not supported for snapshot topics"); + } + + @Test + public void testInitialRetainProgressDoesNotAllowIncrementalRetentionAttributes() { + final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); + final Map<String, String> attributes = newInitialTableTopicAttributes(); + attributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY, "true"); + attributes.put(TopicConstant.RETENTION_BYTES_KEY, "1024"); + assertCreateRejected(subscriptionInfo, attributes, "only supported for incremental topics"); } @Test public void testRejectIllegalRetainProgressValue() { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); - final Map<String, String> attributes = newIncrementalTableTopicAttributes(); - attributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY, "yes"); - - assertCreateRejected(subscriptionInfo, attributes, "expected true or false"); + for (final String mode : + new String[] {TopicConstant.MODE_INITIAL_VALUE, TopicConstant.MODE_INCREMENTAL_VALUE}) { + final Map<String, String> attributes = newIncrementalTableTopicAttributes(); + attributes.put(TopicConstant.MODE_KEY, mode); + attributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY, "yes"); + assertCreateRejected(subscriptionInfo, attributes, "expected true or false"); + } } @Test diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/consumer/DropConsumerProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/consumer/DropConsumerProcedureTest.java index 6d9c16a7dd6..13b9312c558 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/consumer/DropConsumerProcedureTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/consumer/DropConsumerProcedureTest.java @@ -19,23 +19,84 @@ package org.apache.iotdb.confignode.procedure.impl.subscription.consumer; +import org.apache.iotdb.commons.pipe.config.constant.SystemConstant; import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerGroupMeta; import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerMeta; +import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta; +import org.apache.iotdb.confignode.consensus.request.write.subscription.consumer.AlterConsumerGroupPlan; +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.impl.subscription.AbstractOperateSubscriptionProcedure; import org.apache.iotdb.confignode.procedure.store.ProcedureFactory; import org.apache.iotdb.confignode.rpc.thrift.TCloseConsumerReq; +import org.apache.iotdb.rpc.subscription.config.ConsumerConstant; +import org.apache.iotdb.rpc.subscription.config.TopicConstant; import org.apache.tsfile.utils.PublicBAOS; import org.junit.Test; import java.io.DataOutputStream; +import java.lang.reflect.Field; import java.nio.ByteBuffer; +import java.util.Collections; 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.assertTrue; import static org.junit.Assert.fail; public class DropConsumerProcedureTest { + @Test + public void testClosingLastConsumerPersistsConfiguredInitialRuntimeMarker() throws Exception { + for (final boolean retain : new boolean[] {false, true}) { + for (final String dialect : new String[] {"tree", "table"}) { + final SubscriptionInfo subscriptions = new SubscriptionInfo(); + final Map<String, String> attributes = new HashMap<>(); + attributes.put(SystemConstant.SQL_DIALECT_KEY, dialect); + attributes.put(TopicConstant.MODE_KEY, TopicConstant.MODE_INITIAL_VALUE); + attributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY, String.valueOf(retain)); + subscriptions.createTopic(new CreateTopicPlan(new TopicMeta("topic", 1L, attributes))); + final ConsumerGroupMeta group = + new ConsumerGroupMeta( + "group", + 1L, + new ConsumerMeta( + "consumer", + 1L, + Collections.singletonMap(ConsumerConstant.SQL_DIALECT_KEY, dialect))); + group.addSubscription("consumer", Collections.singleton("topic")); + subscriptions.alterConsumerGroup(new AlterConsumerGroupPlan(group)); + final DropConsumerProcedure procedure = + new DropConsumerProcedure(new TCloseConsumerReq("consumer", "group")); + final Field field = + AbstractOperateSubscriptionProcedure.class.getDeclaredField("subscriptionInfo"); + field.setAccessible(true); + field.set(procedure, new AtomicReference<>(subscriptions)); + procedure.executeFromValidate(null); + assertEquals( + retain, + procedure + .getUpdatedConsumerGroupMeta() + .getRetainedInitialTopicNames() + .contains("topic")); + assertTrue(procedure.getUpdatedConsumerGroupMeta().isEmpty()); + + final PublicBAOS bytes = new PublicBAOS(); + procedure.serialize(new DataOutputStream(bytes)); + final ByteBuffer buffer = ByteBuffer.wrap(bytes.getBuf(), 0, bytes.size()); + final DropConsumerProcedure recovered = + (DropConsumerProcedure) ProcedureFactory.getInstance().create(buffer); + assertEquals(procedure, recovered); + assertEquals(0, buffer.remaining()); + subscriptions.alterConsumerGroup( + new AlterConsumerGroupPlan(recovered.getUpdatedConsumerGroupMeta())); + assertEquals(retain, subscriptions.getConsumerGroupMeta("group") != null); + } + } + } + @Test public void serializeDeserializeTest() { PublicBAOS byteArrayOutputStream = new PublicBAOS(); diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/CreateSubscriptionProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/CreateSubscriptionProcedureTest.java index 741e3e344be..047c917e107 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/CreateSubscriptionProcedureTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/CreateSubscriptionProcedureTest.java @@ -20,6 +20,7 @@ package org.apache.iotdb.confignode.procedure.impl.subscription.subscription; import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta; import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerGroupMeta; import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerMeta; import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta; @@ -40,6 +41,7 @@ import org.apache.iotdb.confignode.procedure.store.ProcedureFactory; import org.apache.iotdb.confignode.rpc.thrift.TCreatePipeReq; import org.apache.iotdb.confignode.rpc.thrift.TSubscribeReq; import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.iotdb.rpc.subscription.config.TopicConstant; import org.apache.iotdb.rpc.subscription.exception.SubscriptionException; import org.apache.tsfile.utils.PublicBAOS; @@ -88,6 +90,7 @@ public class CreateSubscriptionProcedureTest { new ConsumerGroupMeta( "test_consumer_group", 1, new ConsumerMeta("old_consumer", 1, consumerAttributes)); newConsumerGroupMeta.addSubscription("old_consumer", subscribeTopics); + newConsumerGroupMeta.retainInitialTopic("retained_topic"); AlterConsumerGroupProcedure alterConsumerGroupProcedure = new AlterConsumerGroupProcedure(newConsumerGroupMeta); @@ -186,6 +189,50 @@ public class CreateSubscriptionProcedureTest { Assert.assertEquals(1, proc.getCreatePipeProcedures().size()); } + @Test + public void executeFromValidateShouldReuseRetainedInitialPipe() throws Exception { + final Map<String, String> consumerAttributes = new HashMap<>(); + consumerAttributes.put("username", "user"); + consumerAttributes.put("password", "password"); + + final ConsumerGroupMeta consumerGroupMeta = + new ConsumerGroupMeta( + "test_consumer_group", 1, new ConsumerMeta("old_consumer", 1, consumerAttributes)); + consumerGroupMeta.retainInitialTopic("test_topic"); + + final Map<String, String> topicAttributes = new HashMap<>(); + topicAttributes.put(TopicConstant.MODE_KEY, TopicConstant.MODE_INITIAL_VALUE); + topicAttributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY, "true"); + final TopicMeta topicMeta = new TopicMeta("test_topic", 1, topicAttributes); + + final SubscriptionInfo subscriptionInfo = Mockito.mock(SubscriptionInfo.class); + Mockito.when(subscriptionInfo.getConsumerGroupMeta("test_consumer_group")) + .thenReturn(consumerGroupMeta); + Mockito.when(subscriptionInfo.deepCopyConsumerGroupMeta("test_consumer_group")) + .thenAnswer(invocation -> consumerGroupMeta.deepCopy()); + Mockito.when(subscriptionInfo.deepCopyTopicMeta("test_topic", false)).thenReturn(topicMeta); + + final PipeTaskInfo pipeTaskInfo = Mockito.mock(PipeTaskInfo.class); + Mockito.when( + pipeTaskInfo.isPipeExisted( + PipeStaticMeta.generateSubscriptionPipeName("test_topic", "test_consumer_group"), + false)) + .thenReturn(true); + + final CreateSubscriptionProcedure proc = + new CreateSubscriptionProcedure( + new TSubscribeReq( + "old_consumer", "test_consumer_group", Collections.singleton("test_topic"))); + setField(proc, "subscriptionInfo", new AtomicReference<>(subscriptionInfo)); + setField(proc, "pipeTaskInfo", new AtomicReference<>(pipeTaskInfo)); + + proc.executeFromValidate(mockCreateSubscriptionValidationEnv()); + + Assert.assertTrue(proc.getCreatePipeProcedures().isEmpty()); + Mockito.verify(pipeTaskInfo, Mockito.never()) + .checkBeforeCreatePipe(Mockito.any(TCreatePipeReq.class)); + } + @Test public void newlySubscribedConsensusTopicsShouldBeBasedOnGroupTopics() { final ConsumerMeta firstConsumer = diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedureTest.java index 1bf74f8db24..d1c64462a5d 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedureTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/subscription/DropSubscriptionProcedureTest.java @@ -33,6 +33,7 @@ import org.apache.iotdb.confignode.procedure.impl.subscription.consumer.AlterCon import org.apache.iotdb.confignode.procedure.store.ProcedureFactory; import org.apache.iotdb.confignode.rpc.thrift.TUnsubscribeReq; import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.iotdb.rpc.subscription.config.TopicConstant; import org.apache.iotdb.rpc.subscription.exception.SubscriptionException; import org.apache.tsfile.utils.PublicBAOS; @@ -165,6 +166,39 @@ public class DropSubscriptionProcedureTest { Assert.assertEquals(1, proc.getDropPipeProcedures().size()); } + @Test + public void executeFromValidateShouldRetainInitialPipeWhenConfigured() throws Exception { + final ConsumerMeta consumerMeta = new ConsumerMeta("old_consumer", 1, Collections.emptyMap()); + final ConsumerGroupMeta consumerGroupMeta = + new ConsumerGroupMeta("test_consumer_group", 1, consumerMeta); + consumerGroupMeta.addSubscription("old_consumer", Collections.singleton("test_topic")); + + final Map<String, String> topicAttributes = new HashMap<>(); + topicAttributes.put(TopicConstant.MODE_KEY, TopicConstant.MODE_INITIAL_VALUE); + topicAttributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY, "true"); + + final SubscriptionInfo subscriptionInfo = Mockito.mock(SubscriptionInfo.class); + Mockito.when(subscriptionInfo.deepCopyConsumerGroupMeta("test_consumer_group")) + .thenAnswer(invocation -> consumerGroupMeta.deepCopy()); + Mockito.when(subscriptionInfo.getConsumerGroupMeta("test_consumer_group")) + .thenReturn(consumerGroupMeta); + Mockito.when(subscriptionInfo.deepCopyTopicMeta("test_topic", false)) + .thenReturn(new TopicMeta("test_topic", 1, topicAttributes)); + + final PipeTaskInfo pipeTaskInfo = Mockito.mock(PipeTaskInfo.class); + final DropSubscriptionProcedure procedure = + new DropSubscriptionProcedure( + new TUnsubscribeReq( + "old_consumer", "test_consumer_group", Collections.singleton("test_topic"))); + setField(procedure, "subscriptionInfo", new AtomicReference<>(subscriptionInfo)); + setField(procedure, "pipeTaskInfo", new AtomicReference<>(pipeTaskInfo)); + + procedure.executeFromValidate(Mockito.mock(ConfigNodeProcedureEnv.class)); + + Assert.assertTrue(procedure.getDropPipeProcedures().isEmpty()); + Mockito.verify(pipeTaskInfo, Mockito.never()).checkBeforeDropPipe(Mockito.anyString()); + } + private static ConfigNodeProcedureEnv mockConsensusFailureEnv(final TSStatus response) throws Exception { final ConfigNodeProcedureEnv env = Mockito.mock(ConfigNodeProcedureEnv.class); 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 56f16b6e14a..e15d7cd7c3c 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 @@ -62,6 +62,12 @@ public class AlterTopicProcedureTest { new TopicMeta("test_topic", 1, topicAttributes), updatedTopicAttributes)); } + @Test + public void serializeDeserializeWithRetainedProgressStateTest() throws Exception { + final AlterTopicProcedure procedure = new AlterTopicProcedure(false, true); + assertSerializeDeserialize(procedure); + } + private void assertSerializeDeserialize(final AlterTopicProcedure procedure) throws Exception { final PublicBAOS byteArrayOutputStream = new PublicBAOS(); final DataOutputStream outputStream = new DataOutputStream(byteArrayOutputStream); @@ -71,6 +77,7 @@ public class AlterTopicProcedureTest { final AlterTopicProcedure deserializedProcedure = (AlterTopicProcedure) ProcedureFactory.getInstance().create(buffer); assertEquals(procedure, deserializedProcedure); + assertEquals(0, buffer.remaining()); } @Test diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/DropTopicProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/DropTopicProcedureTest.java index 0ad614408a3..f210738a31d 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/DropTopicProcedureTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/subscription/topic/DropTopicProcedureTest.java @@ -19,18 +19,216 @@ package org.apache.iotdb.confignode.procedure.impl.subscription.topic; +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.consensus.index.impl.SimpleProgressIndex; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeMeta; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeRuntimeMeta; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStaticMeta; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeStatus; +import org.apache.iotdb.commons.pipe.agent.task.meta.PipeTaskMeta; +import org.apache.iotdb.commons.pipe.config.constant.SystemConstant; +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.ConsumerMeta; +import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta; +import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan; +import org.apache.iotdb.confignode.consensus.request.write.pipe.task.CreatePipePlanV2; +import org.apache.iotdb.confignode.consensus.request.write.pipe.task.OperateMultiplePipesPlanV2; +import org.apache.iotdb.confignode.consensus.request.write.subscription.consumer.AlterConsumerGroupPlan; +import org.apache.iotdb.confignode.consensus.request.write.subscription.consumer.runtime.CommitProgressHandleMetaChangePlan; +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.request.write.subscription.topic.DropTopicPlan; +import org.apache.iotdb.confignode.manager.ConfigManager; +import org.apache.iotdb.confignode.manager.consensus.ConsensusManager; +import org.apache.iotdb.confignode.persistence.pipe.PipeTaskInfo; +import org.apache.iotdb.confignode.persistence.subscription.SubscriptionInfo; +import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv; import org.apache.iotdb.confignode.procedure.store.ProcedureFactory; +import org.apache.iotdb.rpc.TSStatusCode; +import org.apache.iotdb.rpc.subscription.config.ConsumerConstant; +import org.apache.iotdb.rpc.subscription.config.TopicConstant; import org.apache.tsfile.utils.PublicBAOS; import org.junit.Test; +import org.mockito.Mockito; import java.io.DataOutputStream; +import java.lang.reflect.Field; import java.nio.ByteBuffer; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.atomic.AtomicReference; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; public class DropTopicProcedureTest { + + @Test + public void testRetainedRuntimeCleanupAndRollbackAreModelScoped() throws Exception { + for (final boolean dropTopic : new boolean[] {false, true}) { + for (final boolean tableModel : new boolean[] {false, true}) { + assertRetainedRuntimeCleanupAndRollback(dropTopic, tableModel); + } + } + } + + private void assertRetainedRuntimeCleanupAndRollback( + final boolean dropTopic, final boolean tableModel) throws Exception { + final String topicName = "retained_topic"; + final SubscriptionInfo subscriptions = new SubscriptionInfo(); + final PipeTaskInfo pipes = new PipeTaskInfo(); + for (final boolean model : new boolean[] {false, true}) { + final String dialect = model ? "table" : "tree"; + final Map<String, String> attributes = new HashMap<>(); + attributes.put(SystemConstant.SQL_DIALECT_KEY, dialect); + attributes.put(TopicConstant.MODE_KEY, TopicConstant.MODE_INITIAL_VALUE); + attributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY, "true"); + subscriptions.createTopic(new CreateTopicPlan(new TopicMeta(topicName, 1L, attributes))); + final String groupId = "retained_group_" + model; + final ConsumerGroupMeta group = + new ConsumerGroupMeta( + groupId, + 1L, + new ConsumerMeta( + "consumer", + 1L, + Collections.singletonMap(ConsumerConstant.SQL_DIALECT_KEY, dialect))); + group.retainInitialTopic(topicName); + group.removeConsumer("consumer"); + subscriptions.alterConsumerGroup(new AlterConsumerGroupPlan(group)); + final PipeRuntimeMeta runtime = new PipeRuntimeMeta(); + runtime.getStatus().set(PipeStatus.RUNNING); + runtime + .getConsensusGroupId2TaskMetaMap() + .put(1, new PipeTaskMeta(new SimpleProgressIndex(0, 42L), 1)); + pipes.createPipe( + new CreatePipePlanV2( + new PipeStaticMeta( + PipeStaticMeta.generateSubscriptionPipeName(topicName, groupId), + 1L, + Collections.singletonMap(SystemConstant.SQL_DIALECT_KEY, dialect), + Collections.emptyMap(), + Collections.emptyMap()), + runtime)); + final String regionId = "DataRegion[1]"; + subscriptions + .getCommitProgressKeeper() + .updateRegionProgress( + CommitProgressKeeper.generateKey(groupId, topicName, regionId, 1), + ByteBuffer.wrap(new byte[] {(byte) (model ? 2 : 1)})); + subscriptions + .getCommitProgressKeeper() + .updateRegionProgress( + CommitProgressKeeper.generateKey(groupId, "other_topic", regionId, 1), + ByteBuffer.wrap(new byte[] {(byte) (model ? 4 : 3)})); + } + final ConfigNodeProcedureEnv env = Mockito.mock(ConfigNodeProcedureEnv.class); + final ConfigManager manager = Mockito.mock(ConfigManager.class); + final ConsensusManager consensus = Mockito.mock(ConsensusManager.class); + Mockito.when(env.getConfigManager()).thenReturn(manager); + Mockito.when(manager.getConsensusManager()).thenReturn(consensus); + Mockito.when(consensus.write(Mockito.any(ConfigPhysicalPlan.class))) + .thenAnswer( + invocation -> { + final ConfigPhysicalPlan plan = invocation.getArgument(0); + if (plan instanceof CreatePipePlanV2) { + return pipes.createPipe((CreatePipePlanV2) plan); + } else if (plan instanceof OperateMultiplePipesPlanV2) { + return pipes.operateMultiplePipes((OperateMultiplePipesPlanV2) plan); + } else if (plan instanceof AlterConsumerGroupPlan) { + return subscriptions.alterConsumerGroup((AlterConsumerGroupPlan) plan); + } else if (plan instanceof DropTopicPlan) { + return subscriptions.dropTopic((DropTopicPlan) plan); + } else if (plan instanceof CreateTopicPlan) { + return subscriptions.createTopic((CreateTopicPlan) plan); + } else if (plan instanceof AlterTopicPlan) { + return subscriptions.alterTopic((AlterTopicPlan) plan); + } else if (plan instanceof CommitProgressHandleMetaChangePlan) { + return subscriptions.handleCommitProgressChanges( + (CommitProgressHandleMetaChangePlan) plan); + } + return new TSStatus(TSStatusCode.EXECUTE_STATEMENT_ERROR.getStatusCode()); + }); + + final String groupId = "retained_group_" + tableModel; + final String pipeName = PipeStaticMeta.generateSubscriptionPipeName(topicName, groupId); + final String progressKey = + CommitProgressKeeper.generateKey(groupId, topicName, "DataRegion[1]", 1); + final String otherProgressKey = + CommitProgressKeeper.generateKey(groupId, "other_topic", "DataRegion[1]", 1); + final ByteBuffer originalProgress = + subscriptions.getCommitProgressKeeper().getRegionProgress(progressKey); + final ByteBuffer unrelatedProgress = + subscriptions.getCommitProgressKeeper().getRegionProgress(otherProgressKey); + final PipeMeta originalPipe = pipes.getPipeMetaByPipeName(pipeName, tableModel); + final DropTopicProcedure drop = new DropTopicProcedure(topicName, tableModel); + final AlterTopicProcedure alter = + new AlterTopicProcedure( + subscriptions.deepCopyTopicMetaWithUpdatedAttributes( + topicName, + Collections.singletonMap( + TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY, "false"), + tableModel)); + final Object procedure = dropTopic ? drop : alter; + setField(procedure, "subscriptionInfo", new AtomicReference<>(subscriptions)); + setField(procedure, "pipeTaskInfo", new AtomicReference<>(pipes)); + if (dropTopic) { + drop.executeFromValidate(env); + } else { + alter.executeFromValidate(env); + } + final PublicBAOS bytes = new PublicBAOS(); + if (dropTopic) { + drop.serialize(new DataOutputStream(bytes)); + } else { + alter.serialize(new DataOutputStream(bytes)); + } + final ByteBuffer buffer = ByteBuffer.wrap(bytes.getBuf(), 0, bytes.size()); + final Object recovered = ProcedureFactory.getInstance().create(buffer); + assertEquals(0, buffer.remaining()); + assertEquals(procedure, recovered); + setField(recovered, "subscriptionInfo", new AtomicReference<>(subscriptions)); + setField(recovered, "pipeTaskInfo", new AtomicReference<>(pipes)); + if (dropTopic) { + ((DropTopicProcedure) recovered).executeFromOperateOnConfigNodes(env); + } else { + ((AlterTopicProcedure) recovered).executeFromOperateOnConfigNodes(env); + } + assertFalse(pipes.isPipeExisted(pipeName, tableModel)); + assertNull(subscriptions.getCommitProgressKeeper().getRegionProgress(progressKey)); + assertTrue( + subscriptions.getConsumerGroupsRetainingInitialTopic(topicName, tableModel).isEmpty()); + assertEquals( + 1, subscriptions.getConsumerGroupsRetainingInitialTopic(topicName, !tableModel).size()); + if (dropTopic) { + ((DropTopicProcedure) recovered).rollbackFromOperateOnConfigNodes(env); + } else { + ((AlterTopicProcedure) recovered).rollbackFromOperateOnConfigNodes(env); + } + assertEquals( + 1, subscriptions.getConsumerGroupsRetainingInitialTopic(topicName, tableModel).size()); + assertTrue( + subscriptions + .deepCopyTopicMeta(topicName, tableModel) + .getConfig() + .isProgressRetainedAfterUnsubscribe()); + assertEquals( + originalPipe.getRuntimeMeta(), + pipes.getPipeMetaByPipeName(pipeName, tableModel).getRuntimeMeta()); + assertEquals( + originalProgress, subscriptions.getCommitProgressKeeper().getRegionProgress(progressKey)); + assertEquals( + unrelatedProgress, + subscriptions.getCommitProgressKeeper().getRegionProgress(otherProgressKey)); + } + @Test public void serializeDeserializeTest() { PublicBAOS byteArrayOutputStream = new PublicBAOS(); @@ -68,4 +266,50 @@ public class DropTopicProcedureTest { fail(); } } + + @Test + public void serializeDeserializeRetainedProgressStateTest() throws Exception { + final Map<String, String> topicAttributes = new HashMap<>(); + topicAttributes.put(TopicConstant.MODE_KEY, TopicConstant.MODE_INITIAL_VALUE); + topicAttributes.put(TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY, "true"); + final TopicMeta topicMeta = new TopicMeta("test_topic", 1, topicAttributes); + final ConsumerGroupMeta consumerGroupMeta = + new ConsumerGroupMeta( + "test_group", 1, new ConsumerMeta("consumer", 1, Collections.emptyMap())); + consumerGroupMeta.retainInitialTopic("test_topic"); + + final DropTopicProcedure procedure = new DropTopicProcedure("test_topic", false); + setField(procedure, "existedTopicMeta", topicMeta); + final Field retainedConsumerGroupMetasField = + DropTopicProcedure.class.getDeclaredField("retainedConsumerGroupMetas"); + retainedConsumerGroupMetasField.setAccessible(true); + @SuppressWarnings("unchecked") + final List<ConsumerGroupMeta> retainedConsumerGroupMetas = + (List<ConsumerGroupMeta>) retainedConsumerGroupMetasField.get(procedure); + retainedConsumerGroupMetas.add(consumerGroupMeta); + final PublicBAOS byteArrayOutputStream = new PublicBAOS(); + procedure.serialize(new DataOutputStream(byteArrayOutputStream)); + final ByteBuffer buffer = + ByteBuffer.wrap(byteArrayOutputStream.getBuf(), 0, byteArrayOutputStream.size()); + final DropTopicProcedure deserializedProcedure = + (DropTopicProcedure) ProcedureFactory.getInstance().create(buffer); + + assertEquals(procedure, deserializedProcedure); + assertEquals(0, buffer.remaining()); + } + + private static void setField(final Object target, final String fieldName, final Object value) + throws Exception { + for (Class<?> type = target.getClass(); type != null; type = type.getSuperclass()) { + try { + final Field field = type.getDeclaredField(fieldName); + field.setAccessible(true); + field.set(target, value); + return; + } catch (final NoSuchFieldException ignored) { + // Look for inherited procedure state. + } + } + throw new NoSuchFieldException(fieldName); + } } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java index 89d877665eb..cea20c85689 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionBrokerAgent.java @@ -1061,6 +1061,14 @@ public class SubscriptionBrokerAgent { prefetchingQueueCount.invalidate(); } + public void requeuePipeInFlightEvents( + final String consumerGroupId, final String topicName, final String consumerId) { + final SubscriptionBroker broker = getPipeBroker(consumerGroupId); + if (Objects.nonNull(broker)) { + broker.requeueInFlightEvents(topicName, consumerId); + } + } + public boolean executePrefetch(final String consumerGroupId, final String topicName) { final ISubscriptionBroker broker = getBrokerForTopic(consumerGroupId, topicName); if (Objects.isNull(broker)) { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgent.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgent.java index 8af488c5923..74095491e50 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgent.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgent.java @@ -100,13 +100,19 @@ public class SubscriptionConsumerAgent { if (metaFromCoordinator.isEmpty()) { if (Objects.nonNull(metaInAgent)) { - cleanupConsumerGroupSubscriptions(consumerGroupId, metaInAgent, true); + cleanupConsumerGroupSubscriptions( + consumerGroupId, metaInAgent, true, metaFromCoordinator.getRetainedInitialTopicNames()); } final boolean retainProgress = ConsensusSubscriptionSetupHandler.retainConfiguredProgressForConsumerGroup( consumerGroupId, metaFromCoordinator.visibleUnder(true)); - dropConsumerGroupBroker(consumerGroupId); - if (retainProgress) { + final boolean retainInitialProgress = + !metaFromCoordinator.getRetainedInitialTopicNames().isEmpty(); + if (!retainInitialProgress) { + dropConsumerGroupBroker(consumerGroupId); + } + if (retainProgress || retainInitialProgress) { + requeueRetainedInitialEvents(consumerGroupId, metaInAgent, metaFromCoordinator); replaceConsumerGroupMeta(consumerGroupId, metaFromCoordinator); } else { removeConsumerGroupMeta(consumerGroupId); @@ -157,6 +163,11 @@ public class SubscriptionConsumerAgent { // remove prefetching queues for topics unsubscribed by the consumer group final Set<String> topicsUnsubByGroup = ConsumerGroupMeta.getTopicsUnsubByGroup(metaInAgent, metaFromCoordinator); + final Set<String> removedRetainedTopics = + new LinkedHashSet<>(metaInAgent.getRetainedInitialTopicNames()); + removedRetainedTopics.removeAll(metaFromCoordinator.getRetainedInitialTopicNames()); + removedRetainedTopics.removeAll(metaFromCoordinator.getSubscribedTopicNames()); + topicsUnsubByGroup.addAll(removedRetainedTopics); final Set<String> pipeTopicsUnsubByGroup = new LinkedHashSet<>(); final Set<String> consensusTopicsUnsubByGroup = new LinkedHashSet<>(); for (final String topicName : topicsUnsubByGroup) { @@ -165,6 +176,9 @@ public class SubscriptionConsumerAgent { consensusTopicsUnsubByGroup.add(topicName); continue; } + if (metaFromCoordinator.getRetainedInitialTopicNames().contains(topicName)) { + continue; + } pipeTopicsUnsubByGroup.add(topicName); } // Detect newly subscribed topics (present in new meta but not in old meta) @@ -197,6 +211,7 @@ public class SubscriptionConsumerAgent { () -> { // TODO: Currently we fully replace the entire ConsumerGroupMeta without carefully // checking the changes in its fields. + requeueRetainedInitialEvents(consumerGroupId, metaInAgent, metaFromCoordinator); replaceConsumerGroupMeta(consumerGroupId, metaFromCoordinator); }); } @@ -210,6 +225,21 @@ public class SubscriptionConsumerAgent { publishMeta.run(); } + private void requeueRetainedInitialEvents( + final String consumerGroupId, + final ConsumerGroupMeta existingMeta, + final ConsumerGroupMeta updatedMeta) { + if (Objects.isNull(existingMeta)) { + return; + } + for (final String topicName : updatedMeta.getRetainedInitialTopicNames()) { + for (final String consumerId : existingMeta.getConsumersSubscribingTopic(topicName)) { + SubscriptionAgent.broker() + .requeuePipeInFlightEvents(consumerGroupId, topicName, consumerId); + } + } + } + public TPushConsumerGroupMetaRespExceptionMessage handleConsumerGroupMetaChanges( final List<ConsumerGroupMeta> consumerGroupMetasFromCoordinator) { for (final ConsumerGroupMeta consumerGroupMetaFromCoordinator : @@ -349,8 +379,25 @@ public class SubscriptionConsumerAgent { final String consumerGroupId, final ConsumerGroupMeta consumerGroupMeta, final boolean retainConfiguredProgress) { + cleanupConsumerGroupSubscriptions( + consumerGroupId, + consumerGroupMeta, + retainConfiguredProgress, + java.util.Collections.emptySet()); + } + + private void cleanupConsumerGroupSubscriptions( + final String consumerGroupId, + final ConsumerGroupMeta consumerGroupMeta, + final boolean retainConfiguredProgress, + final Set<String> retainedInitialTopicNames) { final Set<String> consensusTopics = new LinkedHashSet<>(); - for (final String topicName : consumerGroupMeta.getSubscribedTopicNames()) { + final Set<String> topicNames = new LinkedHashSet<>(consumerGroupMeta.getSubscribedTopicNames()); + topicNames.addAll(consumerGroupMeta.getRetainedInitialTopicNames()); + for (final String topicName : topicNames) { + if (retainedInitialTopicNames.contains(topicName)) { + continue; + } if (ConsensusSubscriptionSetupHandler.isConsensusBasedTopic( topicName, consumerGroupMeta.visibleUnder(true))) { consensusTopics.add(topicName); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionBroker.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionBroker.java index 3cfb34014d7..5bccf861544 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionBroker.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/SubscriptionBroker.java @@ -402,6 +402,13 @@ public class SubscriptionBroker implements ISubscriptionBroker { .sum(); } + public void requeueInFlightEvents(final String topicName, final String consumerId) { + final SubscriptionPrefetchingQueue queue = topicNameToPrefetchingQueue.get(topicName); + if (Objects.nonNull(queue)) { + queue.requeueInFlightEvents(consumerId); + } + } + @Override public int refreshInFlightEventLeases( final String consumerId, final List<SubscriptionCommitContext> commitContexts) { @@ -491,9 +498,8 @@ public class SubscriptionBroker implements ISubscriptionBroker { public void updateCompletedTopicNames(final String topicName) { // mark topic name completed only for topic of snapshot mode - if (SubscriptionAgent.topic() - .getTopicMode(topicName, isTableModel()) - .equals(TopicConstant.MODE_SNAPSHOT_VALUE)) { + if (TopicConstant.MODE_SNAPSHOT_VALUE.equals( + SubscriptionAgent.topic().getTopicMode(topicName, isTableModel()))) { completedTopicNames.put(topicName, topicName); } } @@ -514,9 +520,8 @@ public class SubscriptionBroker implements ISubscriptionBroker { prefetchingQueue.markClosed(); // mark topic name completed only for topic of snapshot mode - if (SubscriptionAgent.topic() - .getTopicMode(topicName, isTableModel()) - .equals(TopicConstant.MODE_SNAPSHOT_VALUE) + if (TopicConstant.MODE_SNAPSHOT_VALUE.equals( + SubscriptionAgent.topic().getTopicMode(topicName, isTableModel())) && prefetchingQueue.isCompleted()) { completedTopicNames.put(topicName, topicName); } diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgentTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgentTest.java index 0d564ed0c72..100cc78d4cc 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgentTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/subscription/agent/SubscriptionConsumerAgentTest.java @@ -19,12 +19,23 @@ package org.apache.iotdb.db.subscription.agent; +import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerGroupMeta; +import org.apache.iotdb.commons.subscription.meta.consumer.ConsumerMeta; +import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta; +import org.apache.iotdb.db.subscription.broker.ISubscriptionBroker; +import org.apache.iotdb.db.subscription.broker.SubscriptionBroker; +import org.apache.iotdb.db.subscription.broker.SubscriptionPrefetchingQueue; +import org.apache.iotdb.rpc.subscription.config.TopicConstant; + import org.junit.Test; +import org.mockito.Mockito; +import java.lang.reflect.Field; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -33,6 +44,7 @@ import java.util.concurrent.TimeUnit; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import static org.junit.Assert.fail; @@ -40,6 +52,83 @@ public class SubscriptionConsumerAgentTest { private static final long TEST_TIMEOUT_SECONDS = 5L; + @Test + @SuppressWarnings("unchecked") + public void testRetainedInitialQueueSurvivesUnsubscribeAndConsumerClose() throws Exception { + final SubscriptionConsumerAgent agent = SubscriptionAgent.consumer(); + final SubscriptionBrokerAgent brokers = SubscriptionAgent.broker(); + assertNull( + SubscriptionAgent.topic() + .handleSingleTopicMetaChanges( + new TopicMeta( + "topic", + 1L, + Map.of( + TopicConstant.MODE_KEY, + TopicConstant.MODE_INITIAL_VALUE, + TopicConstant.RETAIN_PROGRESS_AFTER_UNSUBSCRIBE_KEY, + "true")))); + try { + for (final boolean closeConsumer : new boolean[] {false, true}) { + final String groupId = "retained_initial_" + closeConsumer; + final ConsumerGroupMeta subscribed = + new ConsumerGroupMeta( + groupId, 1L, new ConsumerMeta("consumer", 1L, Collections.emptyMap())); + subscribed.addSubscription("consumer", Collections.singleton("topic")); + try { + assertNull(agent.handleSingleConsumerGroupMetaChanges(subscribed)); + final Field brokersField = + SubscriptionBrokerAgent.class.getDeclaredField("consumerGroupIdToBrokers"); + brokersField.setAccessible(true); + final Map<String, List<ISubscriptionBroker>> brokerMap = + (Map<String, List<ISubscriptionBroker>>) brokersField.get(brokers); + final SubscriptionBroker broker = (SubscriptionBroker) brokerMap.get(groupId).get(0); + final Field queuesField = + SubscriptionBroker.class.getDeclaredField("topicNameToPrefetchingQueue"); + queuesField.setAccessible(true); + final Map<String, SubscriptionPrefetchingQueue> queues = + (Map<String, SubscriptionPrefetchingQueue>) queuesField.get(broker); + final SubscriptionPrefetchingQueue queue = + Mockito.mock(SubscriptionPrefetchingQueue.class); + Mockito.when(queue.getPrefetchingQueueId()).thenReturn(groupId + "_topic"); + queues.put("topic", queue); + + final ConsumerGroupMeta detached = subscribed.deepCopy(); + detached.removeSubscription("consumer", Collections.singleton("topic")); + detached.retainInitialTopic("topic"); + if (closeConsumer) { + detached.removeConsumer("consumer"); + } + assertNull(agent.handleSingleConsumerGroupMetaChanges(detached)); + assertTrue(brokers.isBrokerExist(groupId)); + assertTrue(broker.hasQueue("topic")); + Mockito.verify(queue).requeueInFlightEvents("consumer"); + Mockito.verify(queue, Mockito.never()).cleanUp(); + + final ConsumerGroupMeta rejoined = detached.deepCopy(); + if (closeConsumer) { + rejoined.addConsumer(new ConsumerMeta("consumer", 2L, Collections.emptyMap())); + } + rejoined.addSubscription("consumer", Collections.singleton("topic")); + rejoined.removeRetainedInitialTopic("topic"); + assertNull(agent.handleSingleConsumerGroupMetaChanges(rejoined)); + assertTrue(broker.hasQueue("topic")); + Mockito.verify(queue, Mockito.never()).cleanUp(); + + final ConsumerGroupMeta cleared = detached.deepCopy(); + cleared.removeRetainedInitialTopic("topic"); + assertNull(agent.handleSingleConsumerGroupMetaChanges(cleared)); + assertFalse(broker.hasQueue("topic")); + Mockito.verify(queue).cleanUp(); + } finally { + assertNull(agent.handleDropConsumerGroup(groupId)); + } + } + } finally { + assertNull(SubscriptionAgent.topic().handleDropTopic("topic", false)); + } + } + @Test public void testTopicDiffSetsUpBeforeTeardownAndMetaPublication() { final List<String> operations = new ArrayList<>(); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/CommitProgressKeeper.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/CommitProgressKeeper.java index 934ef58dd20..e36c4a0633e 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/CommitProgressKeeper.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/CommitProgressKeeper.java @@ -119,6 +119,20 @@ public class CommitProgressKeeper { .anyMatch(prefix -> isValidRegionAndDataNodeProgressKey(key, prefix))); } + public synchronized Map<String, ByteBuffer> getTopicProgress( + final String consumerGroupId, final String topicName) { + final Set<String> prefixes = getTopicProgressKeyPrefixes(consumerGroupId, topicName); + final Map<String, ByteBuffer> result = new HashMap<>(); + regionProgressMap.forEach( + (key, value) -> { + if (prefixes.stream() + .anyMatch(prefix -> isValidRegionAndDataNodeProgressKey(key, prefix))) { + result.put(key, copyBuffer(value)); + } + }); + return result; + } + private static Set<String> getTopicProgressKeyPrefixes( final String consumerGroupId, final String topicName) { final Set<String> prefixes = new HashSet<>(); diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java index a129c5ae70e..7a29d30fc6f 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/subscription/meta/consumer/ConsumerGroupMeta.java @@ -54,6 +54,7 @@ public class ConsumerGroupMeta { private Map<String, ConsumerMeta> consumerIdToConsumerMeta; private boolean tableModel; private Map<String, Long> topicNameToSubscriptionCreationTime; // used when creationTime < 0 + private final Set<String> retainedInitialTopicNames = ConcurrentHashMap.newKeySet(); public ConsumerGroupMeta() { this.topicNameToSubscribedConsumerIdSet = new ConcurrentHashMap<>(); @@ -85,6 +86,7 @@ public class ConsumerGroupMeta { copied.tableModel = tableModel; copied.topicNameToSubscriptionCreationTime = new ConcurrentHashMap<>(topicNameToSubscriptionCreationTime); + copied.retainedInitialTopicNames.addAll(retainedInitialTopicNames); return copied; } @@ -218,6 +220,18 @@ public class ConsumerGroupMeta { return Collections.unmodifiableSet(topicNameToSubscribedConsumerIdSet.keySet()); } + public Set<String> getRetainedInitialTopicNames() { + return Collections.unmodifiableSet(retainedInitialTopicNames); + } + + public void retainInitialTopic(final String topicName) { + retainedInitialTopicNames.add(topicName); + } + + public void removeRetainedInitialTopic(final String topicName) { + retainedInitialTopicNames.remove(topicName); + } + /** * Get the consumers subscribing the given topic in this group. * @@ -338,7 +352,14 @@ public class ConsumerGroupMeta { ReadWriteIOUtils.write(consumerGroupId, outputStream); ReadWriteIOUtils.write(creationTime, outputStream); - ReadWriteIOUtils.write(topicNameToSubscribedConsumerIdSet.size(), outputStream); + // A negative topic count identifies the extended, self-delimiting format. Do not infer the + // presence of retained topics from remaining bytes: this meta is also nested in procedures. + final boolean hasRetainedInitialTopics = !retainedInitialTopicNames.isEmpty(); + ReadWriteIOUtils.write( + hasRetainedInitialTopics + ? -topicNameToSubscribedConsumerIdSet.size() - 1 + : topicNameToSubscribedConsumerIdSet.size(), + outputStream); for (final Map.Entry<String, Set<String>> entry : topicNameToSubscribedConsumerIdSet.entrySet()) { ReadWriteIOUtils.write(entry.getKey(), outputStream); @@ -365,6 +386,12 @@ public class ConsumerGroupMeta { ReadWriteIOUtils.write(entry.getValue(), outputStream); } } + if (hasRetainedInitialTopics) { + ReadWriteIOUtils.write(retainedInitialTopicNames.size(), outputStream); + for (final String topicName : retainedInitialTopicNames) { + ReadWriteIOUtils.write(topicName, outputStream); + } + } } public static ConsumerGroupMeta deserialize(final InputStream inputStream) throws IOException { @@ -375,6 +402,10 @@ public class ConsumerGroupMeta { consumerGroupMeta.topicNameToSubscribedConsumerIdSet = new ConcurrentHashMap<>(); int size = ReadWriteIOUtils.readInt(inputStream); + final boolean hasRetainedInitialTopics = size < 0; + if (hasRetainedInitialTopics) { + size = -size - 1; + } for (int i = 0; i < size; ++i) { final String key = ReadWriteIOUtils.readString(inputStream); @@ -407,6 +438,12 @@ public class ConsumerGroupMeta { } } + if (hasRetainedInitialTopics) { + size = ReadWriteIOUtils.readInt(inputStream); + for (int i = 0; i < size; ++i) { + consumerGroupMeta.retainedInitialTopicNames.add(ReadWriteIOUtils.readString(inputStream)); + } + } return consumerGroupMeta; } @@ -418,6 +455,10 @@ public class ConsumerGroupMeta { consumerGroupMeta.topicNameToSubscribedConsumerIdSet = new ConcurrentHashMap<>(); int size = ReadWriteIOUtils.readInt(byteBuffer); + final boolean hasRetainedInitialTopics = size < 0; + if (hasRetainedInitialTopics) { + size = -size - 1; + } for (int i = 0; i < size; ++i) { final String key = ReadWriteIOUtils.readString(byteBuffer); @@ -450,6 +491,12 @@ public class ConsumerGroupMeta { } } + if (hasRetainedInitialTopics) { + size = ReadWriteIOUtils.readInt(byteBuffer); + for (int i = 0; i < size; ++i) { + consumerGroupMeta.retainedInitialTopicNames.add(ReadWriteIOUtils.readString(byteBuffer)); + } + } return consumerGroupMeta; } @@ -470,6 +517,7 @@ public class ConsumerGroupMeta { && Objects.equals( this.topicNameToSubscribedConsumerIdSet, that.topicNameToSubscribedConsumerIdSet) && Objects.equals(this.consumerIdToConsumerMeta, that.consumerIdToConsumerMeta) + && Objects.equals(this.retainedInitialTopicNames, that.retainedInitialTopicNames) && Objects.equals( this.topicNameToSubscriptionCreationTime, that.topicNameToSubscriptionCreationTime); } @@ -482,6 +530,7 @@ public class ConsumerGroupMeta { tableModel, topicNameToSubscribedConsumerIdSet, consumerIdToConsumerMeta, + retainedInitialTopicNames, topicNameToSubscriptionCreationTime); } @@ -498,6 +547,8 @@ public class ConsumerGroupMeta { + consumerIdToConsumerMeta + ", topicNameToSubscriptionCreationTime=" + topicNameToSubscriptionCreationTime + + ", retainedInitialTopicNames=" + + retainedInitialTopicNames + "}"; } } diff --git a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/consumer/ConsumerGroupDeSerTest.java b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/consumer/ConsumerGroupDeSerTest.java index 2d2014d7e97..6f1cfc3ec5a 100644 --- a/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/consumer/ConsumerGroupDeSerTest.java +++ b/iotdb-core/node-commons/src/test/java/org/apache/iotdb/commons/subscription/consumer/ConsumerGroupDeSerTest.java @@ -29,6 +29,8 @@ import org.junit.Assert; import org.junit.Test; import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.DataOutputStream; import java.io.IOException; import java.nio.ByteBuffer; import java.util.Collections; @@ -39,6 +41,39 @@ import static org.junit.Assert.fail; public class ConsumerGroupDeSerTest { + @Test + public void testRetainedInitialTopicsAreSelfDelimitedAndPreserveModel() throws Exception { + for (final String dialect : new String[] {"tree", "table"}) { + final ConsumerGroupMeta group = + new ConsumerGroupMeta( + "retained", + 1L, + new ConsumerMeta( + "consumer", + 1L, + Collections.singletonMap(ConsumerConstant.SQL_DIALECT_KEY, dialect))); + group.retainInitialTopic("topic"); + group.removeConsumer("consumer"); + final ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + final DataOutputStream output = new DataOutputStream(bytes); + group.serialize(output); + output.writeInt(123456); + + final ByteBuffer buffer = ByteBuffer.wrap(bytes.toByteArray()); + final ConsumerGroupMeta restored = ConsumerGroupMeta.deserialize(buffer); + Assert.assertEquals(group, restored); + Assert.assertEquals(123456, buffer.getInt()); + final ByteArrayInputStream input = new ByteArrayInputStream(bytes.toByteArray()); + Assert.assertEquals(group, ConsumerGroupMeta.deserialize(input)); + Assert.assertEquals(Integer.BYTES, input.available()); + final ConsumerGroupMeta copied = group.deepCopy(); + copied.removeRetainedInitialTopic("topic"); + Assert.assertEquals(Collections.singleton("topic"), group.getRetainedInitialTopicNames()); + Assert.assertTrue(copied.getRetainedInitialTopicNames().isEmpty()); + Assert.assertEquals("table".equals(dialect), restored.visibleUnder(true)); + } + } + @Test public void testEmptyGroupPreservesDialectAcrossSerializationAndRejoin() throws Exception { for (final String dialect : new String[] {"tree", "table"}) {
