This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch to-2011-n in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 8d2e3936da057c473f12cf4f7c0bad99a356d167 Author: Caideyipi <[email protected]> AuthorDate: Mon Aug 3 11:41:25 2026 +0800 Subscription: Rename topic modes to initial and incremental (#18370) (cherry picked from commit d91d236f53db6a327afedb6a1d9a931bb6256d2e) --- .../iotdb/ConsensusSubscriptionSessionExample.java | 2 +- ...sensusTableModelSubscriptionSessionExample.java | 2 +- .../local/ConsensusSubscriptionITSupport.java | 2 +- .../ConsensusSubscriptionTableITSupport.java | 2 +- ...ConsensusSubscriptionColumnFilterClusterIT.java | 2 +- .../IoTDBSubscriptionColumnFilterIT.java | 5 +- .../tablemodel/IoTDBSubscriptionPermissionIT.java | 2 +- .../IoTDBAllTsDatasetPullConsumerIT.java | 2 +- .../IoTDBAllTsTsfilePullConsumerIT.java | 2 +- .../IoTDBPathDeviceDataSetPullConsumerIT.java | 2 +- .../IoTDBPathDeviceTsfilePullConsumerIT.java | 2 +- .../IoTDBTimeTsDatasetPullConsumerIT.java | 2 +- .../IoTDBTimeTsTsfilePullConsumerIT.java | 2 +- .../IoTDBLooseAllTsDatasetPushConsumerIT.java | 4 +- .../IoTDBLooseAllTsfilePushConsumerIT.java | 4 +- .../IoTDBPathLooseDeviceTsfilePushConsumerIT.java | 2 +- .../IoTDBPathLooseTsDatasetPushConsumerIT.java | 2 +- .../IoTDBPathLooseTsfilePushConsumerIT.java | 2 +- .../IoTDBPathTsLooseDatasetPushConsumerIT.java | 2 +- .../IoTDBTimeLooseTsDatasetPushConsumerIT.java | 4 +- .../IoTDBTimeLooseTsTsfilePushConsumerIT.java | 4 +- .../IoTDBTimeLooseTsfilePushConsumerIT.java | 2 +- .../IoTDBTimeTsLooseDatasetPushConsumerIT.java | 4 +- .../subscription/i18n/SubscriptionMessages.java | 2 +- .../subscription/i18n/SubscriptionMessages.java | 2 +- .../iotdb/rpc/subscription/config/TopicConfig.java | 51 ++++++++-- .../rpc/subscription/config/TopicConstant.java | 19 +++- .../base/AbstractSubscriptionConsumer.java | 2 +- .../rpc/subscription/config/TopicConfigTest.java | 57 ++++++++++++ .../iotdb/confignode/i18n/ConfigNodeMessages.java | 7 +- .../iotdb/confignode/i18n/ConfigNodeMessages.java | 7 +- .../runtime/SubscriptionRuntimeCoordinator.java | 2 +- .../persistence/subscription/SubscriptionInfo.java | 44 ++++----- .../SubscriptionHandleLeaderChangeProcedure.java | 2 +- .../subscription/CreateSubscriptionProcedure.java | 2 +- .../subscription/DropSubscriptionProcedure.java | 2 +- .../SubscriptionInfoTopicValidationTest.java | 103 ++++++++++++++------- .../apache/iotdb/db/i18n/DataNodePipeMessages.java | 9 +- .../apache/iotdb/db/i18n/DataNodePipeMessages.java | 9 +- .../config/executor/ClusterConfigTaskExecutor.java | 2 +- .../agent/SubscriptionBrokerAgent.java | 5 +- .../ConsensusSubscriptionSetupHandler.java | 22 ++++- 42 files changed, 285 insertions(+), 123 deletions(-) diff --git a/example/subscription/src/main/java/org/apache/iotdb/ConsensusSubscriptionSessionExample.java b/example/subscription/src/main/java/org/apache/iotdb/ConsensusSubscriptionSessionExample.java index c0ebbe37198..12596f5d5b9 100644 --- a/example/subscription/src/main/java/org/apache/iotdb/ConsensusSubscriptionSessionExample.java +++ b/example/subscription/src/main/java/org/apache/iotdb/ConsensusSubscriptionSessionExample.java @@ -109,7 +109,7 @@ public class ConsensusSubscriptionSessionExample { session.open(); final Properties config = new Properties(); - config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_CONSENSUS_VALUE); + config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_INCREMENTAL_VALUE); config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_RECORD_HANDLER_VALUE); config.put(TopicConstant.PATH_KEY, path); config.put(TopicConstant.ORDER_MODE_KEY, TopicConstant.ORDER_MODE_PER_WRITER_VALUE); diff --git a/example/subscription/src/main/java/org/apache/iotdb/ConsensusTableModelSubscriptionSessionExample.java b/example/subscription/src/main/java/org/apache/iotdb/ConsensusTableModelSubscriptionSessionExample.java index a877a4a861e..fa7733d57f4 100644 --- a/example/subscription/src/main/java/org/apache/iotdb/ConsensusTableModelSubscriptionSessionExample.java +++ b/example/subscription/src/main/java/org/apache/iotdb/ConsensusTableModelSubscriptionSessionExample.java @@ -107,7 +107,7 @@ public class ConsensusTableModelSubscriptionSessionExample { .password(PASSWORD) .build()) { final Properties config = new Properties(); - config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_CONSENSUS_VALUE); + config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_INCREMENTAL_VALUE); config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_RECORD_HANDLER_VALUE); config.put(TopicConstant.DATABASE_KEY, database); config.put(TopicConstant.TABLE_KEY, table); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/ConsensusSubscriptionITSupport.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/ConsensusSubscriptionITSupport.java index 254b5ffeb85..066c38ef831 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/ConsensusSubscriptionITSupport.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/ConsensusSubscriptionITSupport.java @@ -94,7 +94,7 @@ final class ConsensusSubscriptionITSupport { session.dropTopicIfExists(topicName); final Properties config = new Properties(); - config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_CONSENSUS_VALUE); + config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_INCREMENTAL_VALUE); config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_RECORD_HANDLER_VALUE); config.put(TopicConstant.PATH_KEY, path); config.put(TopicConstant.ORDER_MODE_KEY, TopicConstant.ORDER_MODE_PER_WRITER_VALUE); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/ConsensusSubscriptionTableITSupport.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/ConsensusSubscriptionTableITSupport.java index 4bd38992ed1..29a859d3011 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/ConsensusSubscriptionTableITSupport.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/ConsensusSubscriptionTableITSupport.java @@ -134,7 +134,7 @@ final class ConsensusSubscriptionTableITSupport { session.dropTopicIfExists(topicName); final Properties config = new Properties(); - config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_CONSENSUS_VALUE); + config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_INCREMENTAL_VALUE); config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_SESSION_DATA_SETS_HANDLER_VALUE); config.put(TopicConstant.DATABASE_KEY, databasePattern); config.put(TopicConstant.TABLE_KEY, tablePattern); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionColumnFilterClusterIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionColumnFilterClusterIT.java index d2fd5552ee7..6c33532828b 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionColumnFilterClusterIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionColumnFilterClusterIT.java @@ -150,7 +150,7 @@ public class IoTDBConsensusSubscriptionColumnFilterClusterIT extends AbstractSub session.dropTopicIfExists(topicName); final Properties config = new Properties(); - config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_CONSENSUS_VALUE); + config.put(TopicConstant.MODE_KEY, TopicConstant.MODE_INCREMENTAL_VALUE); config.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_SESSION_DATA_SETS_HANDLER_VALUE); config.put(TopicConstant.DATABASE_KEY, database); config.put(TopicConstant.TABLE_KEY, table); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/tablemodel/IoTDBSubscriptionColumnFilterIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/tablemodel/IoTDBSubscriptionColumnFilterIT.java index 27cd10ed534..bfeb7e6b7aa 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/tablemodel/IoTDBSubscriptionColumnFilterIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/dual/tablemodel/IoTDBSubscriptionColumnFilterIT.java @@ -1015,7 +1015,7 @@ public class IoTDBSubscriptionColumnFilterIT extends AbstractSubscriptionDualIT final String columnFilter) throws Exception { createTopic( - topicName, database, tableName, TopicConstant.MODE_LIVE_VALUE, format, columnFilter); + topicName, database, tableName, TopicConstant.MODE_INITIAL_VALUE, format, columnFilter); } private void createTopic( @@ -1032,7 +1032,8 @@ public class IoTDBSubscriptionColumnFilterIT extends AbstractSubscriptionDualIT private void createTopicWithoutColumnFilter( final String topicName, final String database, final String tableName, final String format) throws Exception { - createTopic(topicName, database, tableName, TopicConstant.MODE_LIVE_VALUE, format, "", false); + createTopic( + topicName, database, tableName, TopicConstant.MODE_INITIAL_VALUE, format, "", false); } private void createTopic( diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/tablemodel/IoTDBSubscriptionPermissionIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/tablemodel/IoTDBSubscriptionPermissionIT.java index 123a68c8df5..44d0dea16d7 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/tablemodel/IoTDBSubscriptionPermissionIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/local/tablemodel/IoTDBSubscriptionPermissionIT.java @@ -390,7 +390,7 @@ public class IoTDBSubscriptionPermissionIT extends AbstractSubscriptionLocalIT { private static Properties columnFilterTopicConfig( final String database, final String tableName, final String columnFilter) { final Properties topicConfig = new Properties(); - topicConfig.put(TopicConstant.MODE_KEY, TopicConstant.MODE_LIVE_VALUE); + topicConfig.put(TopicConstant.MODE_KEY, TopicConstant.MODE_INITIAL_VALUE); topicConfig.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_RECORD_HANDLER_VALUE); topicConfig.put(TopicConstant.DATABASE_KEY, database); topicConfig.put(TopicConstant.TABLE_KEY, tableName); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBAllTsDatasetPullConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBAllTsDatasetPullConsumerIT.java index a1a0cae5d9c..fe2ac71aba7 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBAllTsDatasetPullConsumerIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBAllTsDatasetPullConsumerIT.java @@ -75,7 +75,7 @@ public class IoTDBAllTsDatasetPullConsumerIT extends AbstractSubscriptionTreeReg "2024-01-01T00:00:00+08:00", "2024-03-31T23:59:59+08:00", false, - TopicConstant.MODE_LIVE_VALUE, + TopicConstant.MODE_INITIAL_VALUE, TopicConstant.LOOSE_RANGE_ALL_VALUE); session_src.createTimeseries( pattern, TSDataType.INT64, TSEncoding.GORILLA, CompressionType.LZ4); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBAllTsTsfilePullConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBAllTsTsfilePullConsumerIT.java index c00ea2401ce..8a585c5727e 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBAllTsTsfilePullConsumerIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBAllTsTsfilePullConsumerIT.java @@ -78,7 +78,7 @@ public class IoTDBAllTsTsfilePullConsumerIT extends AbstractSubscriptionTreeRegr null, String.valueOf(nowTimestamp), true, - TopicConstant.MODE_LIVE_VALUE, + TopicConstant.MODE_INITIAL_VALUE, TopicConstant.LOOSE_RANGE_ALL_VALUE); session_src.createTimeseries( device + ".s_0", TSDataType.INT64, TSEncoding.GORILLA, CompressionType.LZ4); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBPathDeviceDataSetPullConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBPathDeviceDataSetPullConsumerIT.java index 2cb4b757253..a048041ce90 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBPathDeviceDataSetPullConsumerIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBPathDeviceDataSetPullConsumerIT.java @@ -74,7 +74,7 @@ public class IoTDBPathDeviceDataSetPullConsumerIT extends AbstractSubscriptionTr null, "now", false, - TopicConstant.MODE_LIVE_VALUE, + TopicConstant.MODE_INITIAL_VALUE, TopicConstant.LOOSE_RANGE_PATH_VALUE); session_src.createTimeseries( device + ".s_0", TSDataType.INT64, TSEncoding.GORILLA, CompressionType.LZ4); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBPathDeviceTsfilePullConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBPathDeviceTsfilePullConsumerIT.java index 70a5857ac88..746e2abdcb4 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBPathDeviceTsfilePullConsumerIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBPathDeviceTsfilePullConsumerIT.java @@ -76,7 +76,7 @@ public class IoTDBPathDeviceTsfilePullConsumerIT extends AbstractSubscriptionTre null, null, true, - TopicConstant.MODE_LIVE_VALUE, + TopicConstant.MODE_INITIAL_VALUE, TopicConstant.LOOSE_RANGE_PATH_VALUE); session_src.createTimeseries( device + ".s_0", TSDataType.INT64, TSEncoding.GORILLA, CompressionType.LZ4); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBTimeTsDatasetPullConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBTimeTsDatasetPullConsumerIT.java index 3fc35e39bb9..1b8e9174e9b 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBTimeTsDatasetPullConsumerIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBTimeTsDatasetPullConsumerIT.java @@ -76,7 +76,7 @@ public class IoTDBTimeTsDatasetPullConsumerIT extends AbstractSubscriptionTreeRe "2024-01-01T00:00:00+08:00", "2024-03-31T23:59:59+08:00", false, - TopicConstant.MODE_LIVE_VALUE, + TopicConstant.MODE_INITIAL_VALUE, TopicConstant.LOOSE_RANGE_TIME_VALUE); session_src.createTimeseries( pattern, TSDataType.INT64, TSEncoding.GORILLA, CompressionType.LZ4); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBTimeTsTsfilePullConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBTimeTsTsfilePullConsumerIT.java index 7cec98a6270..f609702e431 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBTimeTsTsfilePullConsumerIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pullconsumer/loose_range/IoTDBTimeTsTsfilePullConsumerIT.java @@ -79,7 +79,7 @@ public class IoTDBTimeTsTsfilePullConsumerIT extends AbstractSubscriptionTreeReg null, String.valueOf(nowTimestamp), true, - TopicConstant.MODE_LIVE_VALUE, + TopicConstant.MODE_INITIAL_VALUE, TopicConstant.LOOSE_RANGE_TIME_VALUE); session_src.createTimeseries( device + ".s_0", TSDataType.INT64, TSEncoding.GORILLA, CompressionType.LZ4); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBLooseAllTsDatasetPushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBLooseAllTsDatasetPushConsumerIT.java index c30e9d8493d..af8d620637f 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBLooseAllTsDatasetPushConsumerIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBLooseAllTsDatasetPushConsumerIT.java @@ -56,7 +56,7 @@ import static org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant.AWAIT * DataSet * pattern: ts * loose-range: all - * mode: live + * mode: initial */ @RunWith(IoTDBTestRunner.class) @Category({MultiClusterIT2SubscriptionTreeRegressionConsumer.class}) @@ -82,7 +82,7 @@ public class IoTDBLooseAllTsDatasetPushConsumerIT extends AbstractSubscriptionTr "2024-01-01T00:00:00+08:00", "2024-02-13T08:00:02+08:00", false, - TopicConstant.MODE_LIVE_VALUE, + TopicConstant.MODE_INITIAL_VALUE, TopicConstant.LOOSE_RANGE_ALL_VALUE); session_src.createTimeseries( device + ".s_0", TSDataType.INT64, TSEncoding.GORILLA, CompressionType.LZ4); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBLooseAllTsfilePushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBLooseAllTsfilePushConsumerIT.java index 91fb1a281e5..1c2f420f339 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBLooseAllTsfilePushConsumerIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBLooseAllTsfilePushConsumerIT.java @@ -58,7 +58,7 @@ import static org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant.AWAIT /*** * push consumer - * mode: live + * mode: initial * pattern: db * loose-range: all */ @@ -84,7 +84,7 @@ public class IoTDBLooseAllTsfilePushConsumerIT extends AbstractSubscriptionTreeR "2024-01-01T00:00:00+08:00", "2024-03-31T00:00:00+08:00", true, - TopicConstant.MODE_LIVE_VALUE, + TopicConstant.MODE_INITIAL_VALUE, TopicConstant.LOOSE_RANGE_ALL_VALUE); session_src.createTimeseries( device + ".s_0", TSDataType.INT64, TSEncoding.GORILLA, CompressionType.LZ4); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseDeviceTsfilePushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseDeviceTsfilePushConsumerIT.java index c647160493a..270d5cd928f 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseDeviceTsfilePushConsumerIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseDeviceTsfilePushConsumerIT.java @@ -83,7 +83,7 @@ public class IoTDBPathLooseDeviceTsfilePushConsumerIT extends AbstractSubscripti "2024-01-01T00:00:00+08:00", "2024-03-31T00:00:00+08:00", true, - TopicConstant.MODE_LIVE_VALUE, + TopicConstant.MODE_INITIAL_VALUE, TopicConstant.LOOSE_RANGE_PATH_VALUE); session_src.createTimeseries( device + ".s_0", TSDataType.INT64, TSEncoding.GORILLA, CompressionType.LZ4); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseTsDatasetPushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseTsDatasetPushConsumerIT.java index 3e144c137fd..de854f29317 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseTsDatasetPushConsumerIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseTsDatasetPushConsumerIT.java @@ -83,7 +83,7 @@ public class IoTDBPathLooseTsDatasetPushConsumerIT extends AbstractSubscriptionT "2024-01-01T00:00:00+08:00", "2024-02-13T08:00:02+08:00", false, - TopicConstant.MODE_LIVE_VALUE, + TopicConstant.MODE_INITIAL_VALUE, TopicConstant.LOOSE_RANGE_PATH_VALUE); session_src.createTimeseries( device + ".s_0", TSDataType.INT64, TSEncoding.GORILLA, CompressionType.LZ4); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseTsfilePushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseTsfilePushConsumerIT.java index b8c438bfb9d..c8800aa93ac 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseTsfilePushConsumerIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseTsfilePushConsumerIT.java @@ -80,7 +80,7 @@ public class IoTDBPathLooseTsfilePushConsumerIT extends AbstractSubscriptionTree "2024-01-01T00:00:00+08:00", "2024-03-31T00:00:00+08:00", true, - TopicConstant.MODE_LIVE_VALUE, + TopicConstant.MODE_INITIAL_VALUE, TopicConstant.LOOSE_RANGE_PATH_VALUE); session_src.createTimeseries( device + ".s_0", TSDataType.INT64, TSEncoding.GORILLA, CompressionType.LZ4); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathTsLooseDatasetPushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathTsLooseDatasetPushConsumerIT.java index 5a2a5d6a351..aba24718a63 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathTsLooseDatasetPushConsumerIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBPathTsLooseDatasetPushConsumerIT.java @@ -81,7 +81,7 @@ public class IoTDBPathTsLooseDatasetPushConsumerIT extends AbstractSubscriptionT "2024-01-01T00:00:00+08:00", "2024-02-13T08:00:02+08:00", false, - TopicConstant.MODE_LIVE_VALUE, + TopicConstant.MODE_INITIAL_VALUE, TopicConstant.LOOSE_RANGE_PATH_VALUE); session_src.createTimeseries( device + ".s_0", TSDataType.INT64, TSEncoding.GORILLA, CompressionType.LZ4); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsDatasetPushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsDatasetPushConsumerIT.java index ecbd661e5e0..38b826f3caf 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsDatasetPushConsumerIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsDatasetPushConsumerIT.java @@ -57,7 +57,7 @@ import static org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant.AWAIT * DataSet * pattern: ts * loose-range: time - * live + * initial */ @RunWith(IoTDBTestRunner.class) @Category({MultiClusterIT2SubscriptionTreeRegressionConsumer.class}) @@ -84,7 +84,7 @@ public class IoTDBTimeLooseTsDatasetPushConsumerIT extends AbstractSubscriptionT "2024-01-01T00:00:00+08:00", "2024-02-13T08:00:02+08:00", false, - TopicConstant.MODE_LIVE_VALUE, + TopicConstant.MODE_INITIAL_VALUE, TopicConstant.LOOSE_RANGE_TIME_VALUE); session_src.createTimeseries( device + ".s_0", TSDataType.INT64, TSEncoding.GORILLA, CompressionType.LZ4); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsTsfilePushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsTsfilePushConsumerIT.java index 0311bc0a0b5..e62644a59dc 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsTsfilePushConsumerIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsTsfilePushConsumerIT.java @@ -57,7 +57,7 @@ import java.util.concurrent.atomic.AtomicInteger; import static org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant.AWAIT; /*** - * mode: live + * mode: initial * loose-range:path * format: tsfile */ @@ -83,7 +83,7 @@ public class IoTDBTimeLooseTsTsfilePushConsumerIT extends AbstractSubscriptionTr "2024-01-01T00:00:00+08:00", "2024-03-31T00:00:00+08:00", true, - TopicConstant.MODE_LIVE_VALUE, + TopicConstant.MODE_INITIAL_VALUE, TopicConstant.LOOSE_RANGE_TIME_VALUE); session_src.createTimeseries( device + ".s_0", TSDataType.INT64, TSEncoding.GORILLA, CompressionType.LZ4); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsfilePushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsfilePushConsumerIT.java index 53207ee7fe6..68b36104b01 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsfilePushConsumerIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsfilePushConsumerIT.java @@ -77,7 +77,7 @@ public class IoTDBTimeLooseTsfilePushConsumerIT extends AbstractSubscriptionTree "2024-01-01T00:00:00+08:00", "2024-03-31T00:00:00+08:00", true, - TopicConstant.MODE_LIVE_VALUE, + TopicConstant.MODE_INITIAL_VALUE, TopicConstant.LOOSE_RANGE_TIME_VALUE); session_src.createTimeseries( device + ".s_0", TSDataType.INT64, TSEncoding.GORILLA, CompressionType.LZ4); diff --git a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeTsLooseDatasetPushConsumerIT.java b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeTsLooseDatasetPushConsumerIT.java index 4f358ddb8c4..bfd539054e9 100644 --- a/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeTsLooseDatasetPushConsumerIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/subscription/it/triple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeTsLooseDatasetPushConsumerIT.java @@ -56,7 +56,7 @@ import static org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant.AWAIT * DataSet * pattern: ts * time loose - * live + * initial */ @RunWith(IoTDBTestRunner.class) @Category({MultiClusterIT2SubscriptionTreeRegressionConsumer.class}) @@ -83,7 +83,7 @@ public class IoTDBTimeTsLooseDatasetPushConsumerIT extends AbstractSubscriptionT "2024-01-01T00:00:00+08:00", "2024-02-13T08:00:02+08:00", false, - TopicConstant.MODE_LIVE_VALUE, + TopicConstant.MODE_INITIAL_VALUE, TopicConstant.LOOSE_RANGE_TIME_VALUE); session_src.createTimeseries( device + ".s_0", TSDataType.INT64, TSEncoding.GORILLA, CompressionType.LZ4); diff --git a/iotdb-client/subscription/src/main/i18n/en/org/apache/iotdb/rpc/subscription/i18n/SubscriptionMessages.java b/iotdb-client/subscription/src/main/i18n/en/org/apache/iotdb/rpc/subscription/i18n/SubscriptionMessages.java index 7ce7e7a8763..95f26d55d68 100644 --- a/iotdb-client/subscription/src/main/i18n/en/org/apache/iotdb/rpc/subscription/i18n/SubscriptionMessages.java +++ b/iotdb-client/subscription/src/main/i18n/en/org/apache/iotdb/rpc/subscription/i18n/SubscriptionMessages.java @@ -270,7 +270,7 @@ public final class SubscriptionMessages { public static final String EXCEPTION_CLUSTER_HAS_NO_AVAILABLE_SUBSCRIPTION_PROVIDERS_ARG_FETCH_ALL_ENDPOINTS_D232693E = "Cluster has no available subscription providers when %s fetch all endpoints"; public static final String EXCEPTION_PASSWORD_ENCRYPTEDPASSWORD_MUTUALLY_EXCLUSIVE_ENCRYPTEDPASSWORD_ALREADY_SET_E4548A43 = "password and encryptedPassword are mutually exclusive; encryptedPassword is already set"; public static final String EXCEPTION_PASSWORD_ENCRYPTEDPASSWORD_MUTUALLY_EXCLUSIVE_PASSWORD_ALREADY_SET_BB20AD1E = "password and encryptedPassword are mutually exclusive; password is already set"; - public static final String EXCEPTION_CONSENSUS_MODE_TOPIC_SHOULD_NOT_GENERATE_PIPE_SOURCE_ATTRIBUTES_BBDFF732 = "Consensus mode topic should not generate pipe source attributes"; + public static final String EXCEPTION_INCREMENTAL_MODE_TOPIC_SHOULD_NOT_GENERATE_PIPE_SOURCE_ATTRIBUTES_09D75393 = "Incremental mode topic should not generate pipe source attributes"; public static final String EXCEPTION_UNSUPPORTED_SUBSCRIPTIONCOMMITCONTEXT_VERSION_8021B27B = "Unsupported SubscriptionCommitContext version: "; public static final String OUTDATED_SUBSCRIPTION_EVENT = "outdated subscription event"; public static final String FIELD_SEPARATOR = ", "; diff --git a/iotdb-client/subscription/src/main/i18n/zh/org/apache/iotdb/rpc/subscription/i18n/SubscriptionMessages.java b/iotdb-client/subscription/src/main/i18n/zh/org/apache/iotdb/rpc/subscription/i18n/SubscriptionMessages.java index 4a8edd3f54a..562a4d72572 100644 --- a/iotdb-client/subscription/src/main/i18n/zh/org/apache/iotdb/rpc/subscription/i18n/SubscriptionMessages.java +++ b/iotdb-client/subscription/src/main/i18n/zh/org/apache/iotdb/rpc/subscription/i18n/SubscriptionMessages.java @@ -219,7 +219,7 @@ public final class SubscriptionMessages { public static final String EXCEPTION_CLUSTER_HAS_NO_AVAILABLE_SUBSCRIPTION_PROVIDERS_ARG_FETCH_ALL_ENDPOINTS_D232693E = "%s 获取所有 endpoint 时，集群没有可用的 SubscriptionProvider"; public static final String EXCEPTION_PASSWORD_ENCRYPTEDPASSWORD_MUTUALLY_EXCLUSIVE_ENCRYPTEDPASSWORD_ALREADY_SET_E4548A43 = "password 与 encryptedPassword 互斥；已设置 encryptedPassword"; public static final String EXCEPTION_PASSWORD_ENCRYPTEDPASSWORD_MUTUALLY_EXCLUSIVE_PASSWORD_ALREADY_SET_BB20AD1E = "password 与 encryptedPassword 互斥；已设置 password"; - public static final String EXCEPTION_CONSENSUS_MODE_TOPIC_SHOULD_NOT_GENERATE_PIPE_SOURCE_ATTRIBUTES_BBDFF732 = "Consensus mode 主题不应生成 pipe source attributes"; + public static final String EXCEPTION_INCREMENTAL_MODE_TOPIC_SHOULD_NOT_GENERATE_PIPE_SOURCE_ATTRIBUTES_09D75393 = "incremental mode 的 topic 不应生成 pipe source attributes"; public static final String EXCEPTION_UNSUPPORTED_SUBSCRIPTIONCOMMITCONTEXT_VERSION_8021B27B = "不支持的 SubscriptionCommitContext 版本："; public static final String OUTDATED_SUBSCRIPTION_EVENT = "过期的订阅事件"; public static final String FIELD_SEPARATOR = "，"; diff --git a/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java b/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java index b0824fcb394..3d107476b85 100644 --- a/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java +++ b/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java @@ -30,6 +30,7 @@ import java.nio.ByteBuffer; import java.util.Collections; import java.util.HashMap; import java.util.HashSet; +import java.util.Locale; import java.util.Map; import java.util.Objects; import java.util.Set; @@ -43,8 +44,8 @@ public class TopicConfig extends PipeParameters { static { final Set<String> modes = new HashSet<>(3); modes.add(TopicConstant.MODE_SNAPSHOT_VALUE); - modes.add(TopicConstant.MODE_LIVE_VALUE); - modes.add(TopicConstant.MODE_CONSENSUS_VALUE); + modes.add(TopicConstant.MODE_INITIAL_VALUE); + modes.add(TopicConstant.MODE_INCREMENTAL_VALUE); MODE_VALUE_SET = Collections.unmodifiableSet(modes); final Set<String> orderModes = new HashSet<>(3); @@ -84,8 +85,10 @@ public class TopicConfig extends PipeParameters { private static final Map<String, String> SNAPSHOT_MODE_CONFIG = Collections.singletonMap("mode", TopicConstant.MODE_SNAPSHOT_VALUE); - private static final Map<String, String> LIVE_MODE_CONFIG = - Collections.singletonMap("mode", TopicConstant.MODE_LIVE_VALUE); + // Pipe source still uses the legacy "live" value for the initial (full + incremental) topic + // mode. + private static final Map<String, String> INITIAL_MODE_CONFIG = + Collections.singletonMap("mode", TopicConstant.LEGACY_MODE_LIVE_VALUE); private static final Map<String, String> STRICT_MODE_CONFIG = Collections.singletonMap("mode.strict", "true"); @@ -125,12 +128,28 @@ public class TopicConfig extends PipeParameters { return TopicConstant.MODE_SNAPSHOT_VALUE.equalsIgnoreCase(getMode()); } + public boolean isInitialMode() { + return TopicConstant.MODE_INITIAL_VALUE.equalsIgnoreCase(getMode()); + } + + public boolean isIncrementalMode() { + return TopicConstant.MODE_INCREMENTAL_VALUE.equalsIgnoreCase(getMode()); + } + + /** + * @deprecated Use {@link #isInitialMode()}. + */ + @Deprecated public boolean isLiveMode() { - return TopicConstant.MODE_LIVE_VALUE.equalsIgnoreCase(getMode()); + return isInitialMode(); } + /** + * @deprecated Use {@link #isIncrementalMode()}. + */ + @Deprecated public boolean isConsensusMode() { - return TopicConstant.MODE_CONSENSUS_VALUE.equalsIgnoreCase(getMode()); + return isIncrementalMode(); } public static boolean isValidMode(final String mode) { @@ -138,7 +157,19 @@ public class TopicConfig extends PipeParameters { } public static String normalizeMode(final String mode) { - return mode == null ? TopicConstant.MODE_DEFAULT_VALUE : mode.trim().toLowerCase(); + if (mode == null) { + return TopicConstant.MODE_DEFAULT_VALUE; + } + + final String normalizedMode = mode.trim().toLowerCase(Locale.ROOT); + switch (normalizedMode) { + case TopicConstant.LEGACY_MODE_LIVE_VALUE: + return TopicConstant.MODE_INITIAL_VALUE; + case TopicConstant.LEGACY_MODE_CONSENSUS_VALUE: + return TopicConstant.MODE_INCREMENTAL_VALUE; + default: + return normalizedMode; + } } public String getOrderMode() { @@ -223,12 +254,12 @@ public class TopicConfig extends PipeParameters { } public Map<String, String> getAttributesWithSourceMode() { - if (isConsensusMode()) { + if (isIncrementalMode()) { throw new IllegalArgumentException( SubscriptionMessages - .EXCEPTION_CONSENSUS_MODE_TOPIC_SHOULD_NOT_GENERATE_PIPE_SOURCE_ATTRIBUTES_BBDFF732); + .EXCEPTION_INCREMENTAL_MODE_TOPIC_SHOULD_NOT_GENERATE_PIPE_SOURCE_ATTRIBUTES_09D75393); } - return isSnapshotMode() ? SNAPSHOT_MODE_CONFIG : LIVE_MODE_CONFIG; + return isSnapshotMode() ? SNAPSHOT_MODE_CONFIG : INITIAL_MODE_CONFIG; } public Map<String, String> getAttributesWithSourceLooseRangeOrStrict() { diff --git a/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java b/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java index f929d7b7472..4d11f2fe8a2 100644 --- a/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java +++ b/iotdb-client/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConstant.java @@ -42,10 +42,23 @@ public class TopicConstant { public static final String NOW_TIME_VALUE = "now"; public static final String MODE_KEY = "mode"; - public static final String MODE_LIVE_VALUE = "live"; + public static final String MODE_INITIAL_VALUE = "initial"; public static final String MODE_SNAPSHOT_VALUE = "snapshot"; - public static final String MODE_CONSENSUS_VALUE = "consensus"; - public static final String MODE_DEFAULT_VALUE = MODE_LIVE_VALUE; + public static final String MODE_INCREMENTAL_VALUE = "incremental"; + public static final String MODE_DEFAULT_VALUE = MODE_INITIAL_VALUE; + + static final String LEGACY_MODE_LIVE_VALUE = "live"; + static final String LEGACY_MODE_CONSENSUS_VALUE = "consensus"; + + /** + * @deprecated Use {@link #MODE_INITIAL_VALUE}. + */ + @Deprecated public static final String MODE_LIVE_VALUE = LEGACY_MODE_LIVE_VALUE; + + /** + * @deprecated Use {@link #MODE_INCREMENTAL_VALUE}. + */ + @Deprecated public static final String MODE_CONSENSUS_VALUE = LEGACY_MODE_CONSENSUS_VALUE; public static final String ORDER_MODE_KEY = "order-mode"; public static final String ORDER_MODE_LEADER_ONLY_VALUE = "leader-only"; diff --git a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java index 74491898e59..fa48b6a099a 100644 --- a/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java +++ b/iotdb-client/subscription/src/main/java/org/apache/iotdb/session/subscription/consumer/base/AbstractSubscriptionConsumer.java @@ -168,7 +168,7 @@ abstract class AbstractSubscriptionConsumer implements AutoCloseable { private boolean allTopicMessagesHaveBeenConsumed(final Collection<String> topicNames) { // For the topic that needs to be detected, there are two scenarios to consider: - // 1. If configs as live, it cannot be determined whether the topic has been fully consumed. + // 1. Initial topics are unbounded and cannot be fully consumed. // 2. If configs as snapshot, it means the topic has not been automatically unsubscribed. // Therefore, the logic can be summarized as follows: if there is a matching topic in subscribed // topics, then it has not been fully consumed. diff --git a/iotdb-client/subscription/src/test/java/org/apache/iotdb/rpc/subscription/config/TopicConfigTest.java b/iotdb-client/subscription/src/test/java/org/apache/iotdb/rpc/subscription/config/TopicConfigTest.java index d60f4f1f702..d308ef123cf 100644 --- a/iotdb-client/subscription/src/test/java/org/apache/iotdb/rpc/subscription/config/TopicConfigTest.java +++ b/iotdb-client/subscription/src/test/java/org/apache/iotdb/rpc/subscription/config/TopicConfigTest.java @@ -28,6 +28,59 @@ import java.util.Map; public class TopicConfigTest { + @Test + public void testModeDefaultsToInitial() { + final TopicConfig topicConfig = new TopicConfig(); + + Assert.assertEquals(TopicConstant.MODE_INITIAL_VALUE, topicConfig.getMode()); + Assert.assertTrue(topicConfig.isInitialMode()); + Assert.assertFalse(topicConfig.isSnapshotMode()); + Assert.assertFalse(topicConfig.isIncrementalMode()); + } + + @Test + public void testCanonicalModeValues() { + Assert.assertTrue(TopicConfig.isValidMode(TopicConstant.MODE_INITIAL_VALUE)); + Assert.assertTrue(TopicConfig.isValidMode(TopicConstant.MODE_SNAPSHOT_VALUE)); + Assert.assertTrue(TopicConfig.isValidMode(TopicConstant.MODE_INCREMENTAL_VALUE)); + Assert.assertFalse(TopicConfig.isValidMode("wal")); + + Assert.assertTrue(topicConfigWithMode(" INITIAL ").isInitialMode()); + Assert.assertTrue(topicConfigWithMode(" INCREMENTAL ").isIncrementalMode()); + } + + @SuppressWarnings("deprecation") + @Test + public void testLegacyModeValues() { + final TopicConfig liveTopicConfig = topicConfigWithMode(TopicConstant.MODE_LIVE_VALUE); + Assert.assertTrue(TopicConfig.isValidMode(TopicConstant.MODE_LIVE_VALUE)); + Assert.assertEquals(TopicConstant.MODE_INITIAL_VALUE, liveTopicConfig.getMode()); + Assert.assertTrue(liveTopicConfig.isInitialMode()); + Assert.assertTrue(liveTopicConfig.isLiveMode()); + + final TopicConfig consensusTopicConfig = + topicConfigWithMode(TopicConstant.MODE_CONSENSUS_VALUE); + Assert.assertTrue(TopicConfig.isValidMode(TopicConstant.MODE_CONSENSUS_VALUE)); + Assert.assertEquals(TopicConstant.MODE_INCREMENTAL_VALUE, consensusTopicConfig.getMode()); + Assert.assertTrue(consensusTopicConfig.isIncrementalMode()); + Assert.assertTrue(consensusTopicConfig.isConsensusMode()); + } + + @SuppressWarnings("deprecation") + @Test + public void testInitialModeMapsToPipeLiveMode() { + Assert.assertEquals( + TopicConstant.MODE_LIVE_VALUE, + topicConfigWithMode(TopicConstant.MODE_INITIAL_VALUE) + .getAttributesWithSourceMode() + .get(TopicConstant.MODE_KEY)); + Assert.assertEquals( + TopicConstant.MODE_LIVE_VALUE, + topicConfigWithMode(TopicConstant.MODE_LIVE_VALUE) + .getAttributesWithSourceMode() + .get(TopicConstant.MODE_KEY)); + } + @Test public void testColumnFilterKeyIsCaseInsensitive() { final TopicConfig topicConfig = @@ -56,4 +109,8 @@ public class TopicConfigTest { Assert.assertTrue(new TopicConfig(attributes).isColumnFilterTrivial()); } + + private static TopicConfig topicConfigWithMode(final String mode) { + return new TopicConfig(Collections.singletonMap(TopicConstant.MODE_KEY, mode)); + } } 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 494201b1e9f..6b21e33317d 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 @@ -663,6 +663,9 @@ public final class ConfigNodeMessages { "procedure_completed_evict_ttl should be greater than 0, but was "; public static final String - EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_MODE_CONSENSUS_DOES_NOT_SUPPORT_TOPIC_ATTRIBUTES_ARG_3C2D0BDA = - "Failed to create or alter topic, mode=consensus does not support topic attributes %s"; + EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_MODE_INCREMENTAL_DOES_NOT_SUPPORT_TOPIC_ATTRIBUTES_ARG_1A72326A = + "Failed to create or alter topic, mode=incremental does not support topic attributes %s"; + public static final String + EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ARG_AND_ARG_ARE_ONLY_SUPPORTED_FOR_INCREMENTAL_TOPICS_D86CEA8E = + "Failed to create or alter topic, %s and %s are only supported for incremental topics"; } 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 0193e5a0485..afd1b24329b 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 @@ -708,6 +708,9 @@ public final class ConfigNodeMessages { EXCEPTION_PROCEDURE_COMPLETED_EVICT_TTL_SHOULD_BE_GREATER_THAN_0_BUT_WAS_5A4D0CF6 = "procedure_completed_evict_ttl 应大于 0，但当前值为 "; public static final String - EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_MODE_CONSENSUS_DOES_NOT_SUPPORT_TOPIC_ATTRIBUTES_ARG_3C2D0BDA = - "创建或修改 topic 失败，mode=consensus 不支持 topic 属性 %s"; + EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_MODE_INCREMENTAL_DOES_NOT_SUPPORT_TOPIC_ATTRIBUTES_ARG_1A72326A = + "创建或修改 topic 失败，mode=incremental 不支持 topic 属性 %s"; + public static final String + EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ARG_AND_ARG_ARE_ONLY_SUPPORTED_FOR_INCREMENTAL_TOPICS_D86CEA8E = + "创建或修改 topic 失败，%s 和 %s 仅支持 incremental 模式的 topic"; } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/runtime/SubscriptionRuntimeCoordinator.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/runtime/SubscriptionRuntimeCoordinator.java index 1f3a70073c5..dd8fe2e70cf 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/runtime/SubscriptionRuntimeCoordinator.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/runtime/SubscriptionRuntimeCoordinator.java @@ -102,7 +102,7 @@ public class SubscriptionRuntimeCoordinator { .getSubscriptionCoordinator() .getSubscriptionInfo() .getAllTopicMeta()) { - if (topicMeta.getConfig().isConsensusMode()) { + if (topicMeta.getConfig().isIncrementalMode()) { return true; } } 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 a9686652465..d472758d1fb 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 @@ -110,7 +110,7 @@ public class SubscriptionInfo implements SnapshotProcessor { TopicConstant.OWNER_EPOCH_KEY, TopicConstant.MAX_OWNER_EPOCH_KEY, TopicConstant.OWNER_LEASE_DURATION_MS_KEY); - private static final Set<String> CONSENSUS_TOPIC_SUPPORTED_ATTRIBUTE_KEYS = + private static final Set<String> INCREMENTAL_TOPIC_SUPPORTED_ATTRIBUTE_KEYS = Set.of( SystemConstant.SQL_DIALECT_KEY, TopicConstant.PATH_KEY, @@ -340,21 +340,21 @@ public class SubscriptionInfo implements SnapshotProcessor { TopicConstant.MODE_KEY, mode, TopicConstant.MODE_SNAPSHOT_VALUE, - TopicConstant.MODE_LIVE_VALUE, - TopicConstant.MODE_CONSENSUS_VALUE); + TopicConstant.MODE_INITIAL_VALUE, + TopicConstant.MODE_INCREMENTAL_VALUE); LOGGER.warn(exceptionMessage); throw new SubscriptionException(exceptionMessage); } - validateConsensusTopicAttributes(topicConfig); - validateConsensusProtocolSupport(topicConfig); + validateIncrementalTopicAttributes(topicConfig); + validateIncrementalProtocolSupport(topicConfig); - if (topicConfig.isConsensusMode() && !topicConfig.isRecordFormat()) { + if (topicConfig.isIncrementalMode() && !topicConfig.isRecordFormat()) { final String exceptionMessage = String.format( "Failed to create or alter topic, %s=%s only supports %s=%s", TopicConstant.MODE_KEY, - TopicConstant.MODE_CONSENSUS_VALUE, + TopicConstant.MODE_INCREMENTAL_VALUE, TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_RECORD_HANDLER_VALUE); LOGGER.warn(exceptionMessage); @@ -376,7 +376,7 @@ public class SubscriptionInfo implements SnapshotProcessor { } validateColumnFilter(topicConfig); - validateConsensusTopicRetentionConfig(topicConfig); + validateIncrementalTopicRetentionConfig(topicConfig); final Long ownerLeaseDurationMs = topicConfig.getLong(TopicConstant.OWNER_LEASE_DURATION_MS_KEY); @@ -393,9 +393,9 @@ public class SubscriptionInfo implements SnapshotProcessor { } } - private void validateConsensusTopicAttributes(final TopicConfig topicConfig) + private void validateIncrementalTopicAttributes(final TopicConfig topicConfig) throws SubscriptionException { - if (!topicConfig.isConsensusMode()) { + if (!topicConfig.isIncrementalMode()) { return; } @@ -404,7 +404,7 @@ public class SubscriptionInfo implements SnapshotProcessor { .filter( key -> Objects.isNull(key) - || !CONSENSUS_TOPIC_SUPPORTED_ATTRIBUTE_KEYS.contains( + || !INCREMENTAL_TOPIC_SUPPORTED_ATTRIBUTE_KEYS.contains( key.trim().toLowerCase(Locale.ROOT))) .map(String::valueOf) .sorted() @@ -416,15 +416,15 @@ public class SubscriptionInfo implements SnapshotProcessor { final String exceptionMessage = String.format( ConfigNodeMessages - .EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_MODE_CONSENSUS_DOES_NOT_SUPPORT_TOPIC_ATTRIBUTES_ARG_3C2D0BDA, + .EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_MODE_INCREMENTAL_DOES_NOT_SUPPORT_TOPIC_ATTRIBUTES_ARG_1A72326A, unsupportedAttributes); LOGGER.warn(exceptionMessage); throw new SubscriptionException(exceptionMessage); } - private void validateConsensusProtocolSupport(final TopicConfig topicConfig) + private void validateIncrementalProtocolSupport(final TopicConfig topicConfig) throws SubscriptionException { - if (!topicConfig.isConsensusMode()) { + if (!topicConfig.isIncrementalMode()) { return; } @@ -437,7 +437,7 @@ public class SubscriptionInfo implements SnapshotProcessor { String.format( "Failed to create or alter topic, %s=%s is only supported when %s=%s, but current value is %s", TopicConstant.MODE_KEY, - TopicConstant.MODE_CONSENSUS_VALUE, + TopicConstant.MODE_INCREMENTAL_VALUE, DATA_REGION_CONSENSUS_PROTOCOL_CLASS_KEY, ConsensusFactory.IOT_CONSENSUS, actualProtocol); @@ -491,22 +491,24 @@ public class SubscriptionInfo implements SnapshotProcessor { } } - private boolean isConsensusBasedTopicConfig(final TopicConfig topicConfig) { - return topicConfig.isConsensusMode(); + private boolean isIncrementalTopicConfig(final TopicConfig topicConfig) { + return topicConfig.isIncrementalMode(); } - private void validateConsensusTopicRetentionConfig(final TopicConfig topicConfig) + private void validateIncrementalTopicRetentionConfig(final TopicConfig topicConfig) throws SubscriptionException { if (!topicConfig.hasAttribute(TopicConstant.RETENTION_BYTES_KEY) && !topicConfig.hasAttribute(TopicConstant.RETENTION_MS_KEY)) { return; } - if (!isConsensusBasedTopicConfig(topicConfig)) { + if (!isIncrementalTopicConfig(topicConfig)) { final String exceptionMessage = String.format( - "Failed to create or alter topic, %s and %s are only supported for consensus topics", - TopicConstant.RETENTION_BYTES_KEY, TopicConstant.RETENTION_MS_KEY); + ConfigNodeMessages + .EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_ARG_AND_ARG_ARE_ONLY_SUPPORTED_FOR_INCREMENTAL_TOPICS_D86CEA8E, + TopicConstant.RETENTION_BYTES_KEY, + TopicConstant.RETENTION_MS_KEY); LOGGER.warn(exceptionMessage); throw new SubscriptionException(exceptionMessage); } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/runtime/SubscriptionHandleLeaderChangeProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/runtime/SubscriptionHandleLeaderChangeProcedure.java index 9c0801b0f86..c1af7445de3 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/runtime/SubscriptionHandleLeaderChangeProcedure.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/subscription/runtime/SubscriptionHandleLeaderChangeProcedure.java @@ -94,7 +94,7 @@ public class SubscriptionHandleLeaderChangeProcedure extends AbstractOperateSubs return false; } for (final TopicMeta topicMeta : subscriptionInfo.get().getAllTopicMeta()) { - if (topicMeta.getConfig().isConsensusMode()) { + if (topicMeta.getConfig().isIncrementalMode()) { 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 1d8d5bc22b4..71792c71c0b 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 @@ -116,7 +116,7 @@ public class CreateSubscriptionProcedure extends AbstractOperateSubscriptionAndP final TopicMeta topicMeta = subscriptionInfo.get().deepCopyTopicMeta(topicName); final String topicMode = topicMeta.getConfig().getMode(); - final boolean isConsensusBasedTopic = topicMeta.getConfig().isConsensusMode(); + final boolean isConsensusBasedTopic = topicMeta.getConfig().isIncrementalMode(); if (isConsensusBasedTopic) { // skip pipe creation 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 2329a81d83d..321ebc9bb96 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 @@ -108,7 +108,7 @@ public class DropSubscriptionProcedure extends AbstractOperateSubscriptionAndPip if (topicsUnsubByGroup.contains(topic)) { final TopicMeta topicMeta = subscriptionInfo.get().deepCopyTopicMeta(topic); final String topicMode = topicMeta.getConfig().getMode(); - final boolean isConsensusBasedTopic = topicMeta.getConfig().isConsensusMode(); + final boolean isConsensusBasedTopic = topicMeta.getConfig().isIncrementalMode(); if (isConsensusBasedTopic) { LOGGER.info( 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 858bc41d758..bda9cf81d7f 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 @@ -38,7 +38,7 @@ public class SubscriptionInfoTopicValidationTest { @Test public void testValidateColumnFilterOnCreate() throws Exception { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); - final Map<String, String> attributes = newConsensusTableTopicAttributes(); + final Map<String, String> attributes = newIncrementalTableTopicAttributes(); attributes.put(TopicConstant.COLUMN_FILTER_KEY, "column_name IN (\"id1\", \"m1\")"); Assert.assertTrue( @@ -58,7 +58,7 @@ public class SubscriptionInfoTopicValidationTest { @Test public void testColumnFilterKeyIsCaseInsensitiveOnCreate() throws Exception { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); - final Map<String, String> attributes = newLiveTableTopicAttributes(); + final Map<String, String> attributes = newInitialTableTopicAttributes(); attributes.put("Column-Filter", "column_name = \"id1\""); Assert.assertTrue( @@ -69,7 +69,7 @@ public class SubscriptionInfoTopicValidationTest { @Test public void testRejectDuplicateColumnFilterKeys() { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); - final Map<String, String> attributes = newLiveTableTopicAttributes(); + final Map<String, String> attributes = newInitialTableTopicAttributes(); attributes.put(TopicConstant.COLUMN_FILTER_KEY, "column_name = \"id1\""); attributes.put("Column-Filter", "column_name = \"m1\""); @@ -88,16 +88,16 @@ public class SubscriptionInfoTopicValidationTest { @Test public void testRejectDuplicateTopicConfigKeys() { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); - final Map<String, String> attributes = newLiveTableTopicAttributes(); + final Map<String, String> attributes = newInitialTableTopicAttributes(); attributes.put("Mode", TopicConstant.MODE_SNAPSHOT_VALUE); assertCreateRejected(subscriptionInfo, attributes, "duplicate mode"); } @Test - public void testAcceptColumnFilterOnLiveTsFileTableTopic() throws Exception { + public void testAcceptColumnFilterOnInitialTsFileTableTopic() throws Exception { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); - final Map<String, String> attributes = newLiveTableTopicAttributes(); + final Map<String, String> attributes = newInitialTableTopicAttributes(); attributes.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_TS_FILE_VALUE); attributes.put(TopicConstant.COLUMN_FILTER_KEY, "column_name = \"id1\""); @@ -107,18 +107,18 @@ public class SubscriptionInfoTopicValidationTest { } @Test - public void testRejectLegacyTsFileAliasOnConsensusTopic() { + public void testRejectLegacyTsFileAliasOnIncrementalTopic() { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); - final Map<String, String> attributes = newConsensusTableTopicAttributes(); + final Map<String, String> attributes = newIncrementalTableTopicAttributes(); attributes.put(TopicConstant.FORMAT_KEY, "TsFileHandler"); - assertCreateRejected(subscriptionInfo, attributes, "mode=consensus only supports format"); + assertCreateRejected(subscriptionInfo, attributes, "mode=incremental only supports format"); } @Test - public void testRejectUnsupportedAttributesOnConsensusTopic() { + public void testRejectUnsupportedAttributesOnIncrementalTopic() { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); - final Map<String, String> attributes = newConsensusTableTopicAttributes(); + final Map<String, String> attributes = newIncrementalTableTopicAttributes(); attributes.put(TopicConstant.START_TIME_KEY, "0"); attributes.put(TopicConstant.STRICT_KEY, "false"); attributes.put("processor", "custom-processor"); @@ -126,25 +126,25 @@ public class SubscriptionInfoTopicValidationTest { assertCreateRejected( subscriptionInfo, attributes, - "mode=consensus does not support topic attributes [processor, start-time, strict]"); + "mode=incremental does not support topic attributes [processor, start-time, strict]"); } @Test - public void testRejectUnknownAttributeOnConsensusTopic() { + public void testRejectUnknownAttributeOnIncrementalTopic() { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); - final Map<String, String> attributes = newConsensusTableTopicAttributes(); + final Map<String, String> attributes = newIncrementalTableTopicAttributes(); attributes.put("unknown-attribute", "value"); assertCreateRejected( subscriptionInfo, attributes, - "mode=consensus does not support topic attributes [unknown-attribute]"); + "mode=incremental does not support topic attributes [unknown-attribute]"); } @Test - public void testAllowPipeAttributesOnLiveTopic() throws Exception { + public void testAllowPipeAttributesOnInitialTopic() throws Exception { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); - final Map<String, String> attributes = newLiveTableTopicAttributes(); + final Map<String, String> attributes = newInitialTableTopicAttributes(); attributes.put(TopicConstant.START_TIME_KEY, "0"); attributes.put(TopicConstant.STRICT_KEY, "false"); attributes.put("processor", "custom-processor"); @@ -157,7 +157,7 @@ public class SubscriptionInfoTopicValidationTest { @Test public void testRejectEmptyColumnFilter() { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); - final Map<String, String> attributes = newConsensusTableTopicAttributes(); + final Map<String, String> attributes = newIncrementalTableTopicAttributes(); attributes.put(TopicConstant.COLUMN_FILTER_KEY, " "); assertCreateRejected(subscriptionInfo, attributes, "column-filter should not be empty"); @@ -166,12 +166,12 @@ public class SubscriptionInfoTopicValidationTest { @Test public void testAcceptAlteringColumnFilter() throws Exception { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); - final Map<String, String> originalAttributes = newConsensusTableTopicAttributes(); + final Map<String, String> originalAttributes = newIncrementalTableTopicAttributes(); originalAttributes.put(TopicConstant.COLUMN_FILTER_KEY, "column_name = \"id1\""); subscriptionInfo.createTopic( new CreateTopicPlan(new TopicMeta("table_topic", 1L, originalAttributes))); - final Map<String, String> updatedAttributes = newConsensusTableTopicAttributes(); + final Map<String, String> updatedAttributes = newIncrementalTableTopicAttributes(); updatedAttributes.put(TopicConstant.COLUMN_FILTER_KEY, "column_name = \"m1\""); subscriptionInfo.validateBeforeAlteringTopic( @@ -181,7 +181,7 @@ public class SubscriptionInfoTopicValidationTest { @Test public void testValidateRetentionConfigOnCreate() throws Exception { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); - final Map<String, String> attributes = newConsensusTableTopicAttributes(); + final Map<String, String> attributes = newIncrementalTableTopicAttributes(); attributes.put(TopicConstant.RETENTION_BYTES_KEY, "1048576"); attributes.put(TopicConstant.RETENTION_MS_KEY, "-1"); @@ -193,17 +193,17 @@ public class SubscriptionInfoTopicValidationTest { @Test public void testRejectRetentionOnTsFileTopic() { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); - final Map<String, String> attributes = newConsensusTableTopicAttributes(); + final Map<String, String> attributes = newIncrementalTableTopicAttributes(); attributes.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_TS_FILE_VALUE); attributes.put(TopicConstant.RETENTION_BYTES_KEY, "1024"); - assertCreateRejected(subscriptionInfo, attributes, "mode=consensus only supports format"); + assertCreateRejected(subscriptionInfo, attributes, "mode=incremental only supports format"); } @Test public void testRejectIllegalRetentionValue() { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); - final Map<String, String> attributes = newConsensusTableTopicAttributes(); + final Map<String, String> attributes = newIncrementalTableTopicAttributes(); attributes.put(TopicConstant.RETENTION_BYTES_KEY, "0"); assertCreateRejected(subscriptionInfo, attributes, "expected -1 or a positive long value"); @@ -212,7 +212,7 @@ public class SubscriptionInfoTopicValidationTest { @Test public void testRejectIllegalRetentionFormat() { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); - final Map<String, String> attributes = newConsensusTableTopicAttributes(); + final Map<String, String> attributes = newIncrementalTableTopicAttributes(); attributes.put(TopicConstant.RETENTION_MS_KEY, "1h"); assertCreateRejected(subscriptionInfo, attributes, "expected a long value"); @@ -221,12 +221,12 @@ public class SubscriptionInfoTopicValidationTest { @Test public void testRejectAlteringRetentionConfig() throws Exception { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); - final Map<String, String> originalAttributes = newConsensusTableTopicAttributes(); + final Map<String, String> originalAttributes = newIncrementalTableTopicAttributes(); originalAttributes.put(TopicConstant.RETENTION_BYTES_KEY, "1024"); subscriptionInfo.createTopic( new CreateTopicPlan(new TopicMeta("table_topic", 1L, originalAttributes))); - final Map<String, String> updatedAttributes = newConsensusTableTopicAttributes(); + final Map<String, String> updatedAttributes = newIncrementalTableTopicAttributes(); updatedAttributes.put(TopicConstant.RETENTION_BYTES_KEY, "2048"); try { @@ -247,10 +247,41 @@ public class SubscriptionInfoTopicValidationTest { assertCreateRejected(subscriptionInfo, attributes, "unsupported mode"); } + @SuppressWarnings("deprecation") @Test - public void testAcceptColumnFilterOnLiveTableTopic() throws Exception { + public void testAcceptLegacyModeValues() throws Exception { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); - final Map<String, String> attributes = newLiveTableTopicAttributes(); + + final Map<String, String> liveAttributes = newInitialTableTopicAttributes(); + liveAttributes.put(TopicConstant.MODE_KEY, TopicConstant.MODE_LIVE_VALUE); + Assert.assertTrue( + subscriptionInfo.validateBeforeCreatingTopic( + new TCreateTopicReq("live_topic").setTopicAttributes(liveAttributes))); + + final Map<String, String> consensusAttributes = newIncrementalTableTopicAttributes(); + consensusAttributes.put(TopicConstant.MODE_KEY, TopicConstant.MODE_CONSENSUS_VALUE); + Assert.assertTrue( + subscriptionInfo.validateBeforeCreatingTopic( + new TCreateTopicReq("consensus_topic").setTopicAttributes(consensusAttributes))); + } + + @SuppressWarnings("deprecation") + @Test + public void testAllowAlteringModeFromLegacyAliasToCanonicalValue() throws Exception { + final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); + final Map<String, String> originalAttributes = newInitialTableTopicAttributes(); + originalAttributes.put(TopicConstant.MODE_KEY, TopicConstant.MODE_LIVE_VALUE); + subscriptionInfo.createTopic( + new CreateTopicPlan(new TopicMeta("table_topic", 1L, originalAttributes))); + + subscriptionInfo.validateBeforeAlteringTopic( + new TopicMeta("table_topic", 2L, newInitialTableTopicAttributes())); + } + + @Test + public void testAcceptColumnFilterOnInitialTableTopic() throws Exception { + final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); + final Map<String, String> attributes = newInitialTableTopicAttributes(); attributes.put(TopicConstant.COLUMN_FILTER_KEY, "column_name = \"id1\""); Assert.assertTrue( @@ -259,12 +290,12 @@ public class SubscriptionInfoTopicValidationTest { } @Test - public void testRejectConsensusOnlyRetentionOnLiveTopic() { + public void testRejectIncrementalOnlyRetentionOnInitialTopic() { final SubscriptionInfo subscriptionInfo = new SubscriptionInfo(); - final Map<String, String> attributes = newLiveTableTopicAttributes(); + final Map<String, String> attributes = newInitialTableTopicAttributes(); attributes.put(TopicConstant.RETENTION_BYTES_KEY, "1024"); - assertCreateRejected(subscriptionInfo, attributes, "only supported for consensus topics"); + assertCreateRejected(subscriptionInfo, attributes, "only supported for incremental topics"); } @Test @@ -294,18 +325,18 @@ public class SubscriptionInfoTopicValidationTest { new TCreateTopicReq("owner_topic").setTopicAttributes(attributes))); } - private static Map<String, String> newConsensusTableTopicAttributes() { + private static Map<String, String> newIncrementalTableTopicAttributes() { final Map<String, String> attributes = new HashMap<>(); attributes.put(SystemConstant.SQL_DIALECT_KEY, SystemConstant.SQL_DIALECT_TABLE_VALUE); - attributes.put(TopicConstant.MODE_KEY, TopicConstant.MODE_CONSENSUS_VALUE); + attributes.put(TopicConstant.MODE_KEY, TopicConstant.MODE_INCREMENTAL_VALUE); attributes.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_RECORD_HANDLER_VALUE); return attributes; } - private static Map<String, String> newLiveTableTopicAttributes() { + private static Map<String, String> newInitialTableTopicAttributes() { final Map<String, String> attributes = new HashMap<>(); attributes.put(SystemConstant.SQL_DIALECT_KEY, SystemConstant.SQL_DIALECT_TABLE_VALUE); - attributes.put(TopicConstant.MODE_KEY, TopicConstant.MODE_LIVE_VALUE); + attributes.put(TopicConstant.MODE_KEY, TopicConstant.MODE_INITIAL_VALUE); attributes.put(TopicConstant.FORMAT_KEY, TopicConstant.FORMAT_RECORD_HANDLER_VALUE); return attributes; } diff --git a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 062d11fc5c7..62c0e359362 100644 --- a/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/en/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -2009,10 +2009,15 @@ public final class DataNodePipeMessages { + "runtimeVersion {} -> {}, runtimeState={} (route hint)"; public static final String PIPE_LOG_FAILED_TO_CHECK_IF_TOPIC_IS_CONSENSUS_BASED_DEFAULTING_TO_ECCE1509 = "Failed to check if topic [{}] is consensus-based, defaulting to false"; - public static final String PIPE_LOG_SKIPPING_SETUP_OF_CONSENSUS_BASED_SUBSCRIPTIONS_FOR_CONSUMER_A7B2C812 = + public static final String PIPE_LOG_SKIPPING_SETUP_OF_CONSENSUS_BASED_SUBSCRIPTIONS_FOR_CONSUMER_46BEE6E4 = "Skipping setup of consensus-based subscriptions for consumer group [{}] because " - + "mode=consensus only supports data_region_consensus_protocol_class={}, but current " + + "mode=incremental only supports data_region_consensus_protocol_class={}, but current " + "configured value is {} (runtime consensus implementation: {})"; + public static final String + EXCEPTION_SUBSCRIPTION_CANNOT_ARG_CONSENSUS_BASED_TOPIC_S_ARG_IN_CONSUMER_GROUP_ARG_BECAUSE_MODE_INCREMENTAL_ONLY_SUPPORTS_DATA_REGION_CONSENSUS_PROTOCOL_CLASS_ARG_BUT_CURRENT_CONFIGURED_VALUE_IS_ARG_RUNTIME_CONSENSUS_IMPLEMENTATION_ARG_6F21ED67 = + "Subscription: cannot %s consensus-based topic(s) %s in consumer group [%s] because " + + "mode=incremental only supports data_region_consensus_protocol_class=%s, but " + + "current configured value is %s (runtime consensus implementation: %s)"; public static final String PIPE_LOG_TOPIC_CONFIG_NOT_FOUND_FOR_TOPIC_CANNOT_SET_UP_CONSENSUS_A93339CE = "Topic config not found for topic [{}], cannot set up consensus queue"; public static final String PIPE_LOG_NO_LOCAL_IOTCONSENSUS_DATA_REGION_FOUND_FOR_TOPIC_IN_CONSUMER_6FD0600E = diff --git a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java index 7da320584cb..9f7ce3ad370 100644 --- a/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java +++ b/iotdb-core/datanode/src/main/i18n/zh/org/apache/iotdb/db/i18n/DataNodePipeMessages.java @@ -1865,9 +1865,14 @@ public final class DataNodePipeMessages { + "{} -> {}，runtimeState={}（route hint）"; public static final String PIPE_LOG_FAILED_TO_CHECK_IF_TOPIC_IS_CONSENSUS_BASED_DEFAULTING_TO_ECCE1509 = "检查 topic [{}] 是否为 consensus-based 失败，默认设为 false"; - public static final String PIPE_LOG_SKIPPING_SETUP_OF_CONSENSUS_BASED_SUBSCRIPTIONS_FOR_CONSUMER_A7B2C812 = - "跳过 consumer group [{}] 的 consensus-based subscription 设置，因为 mode=consensus 仅支持 " + public static final String PIPE_LOG_SKIPPING_SETUP_OF_CONSENSUS_BASED_SUBSCRIPTIONS_FOR_CONSUMER_46BEE6E4 = + "跳过 consumer group [{}] 的 consensus-based subscription 设置，因为 mode=incremental 仅支持 " + "data_region_consensus_protocol_class={}，但当前配置值为 {}（运行时 consensus 实现：{}）"; + public static final String + EXCEPTION_SUBSCRIPTION_CANNOT_ARG_CONSENSUS_BASED_TOPIC_S_ARG_IN_CONSUMER_GROUP_ARG_BECAUSE_MODE_INCREMENTAL_ONLY_SUPPORTS_DATA_REGION_CONSENSUS_PROTOCOL_CLASS_ARG_BUT_CURRENT_CONFIGURED_VALUE_IS_ARG_RUNTIME_CONSENSUS_IMPLEMENTATION_ARG_6F21ED67 = + "Subscription：无法执行 %s，consensus-based topic 为 %s，consumer group 为 [%s]，因为 " + + "mode=incremental 仅支持 data_region_consensus_protocol_class=%s，但当前配置值为 %s" + + "（运行时 consensus 实现：%s）"; public static final String PIPE_LOG_TOPIC_CONFIG_NOT_FOUND_FOR_TOPIC_CANNOT_SET_UP_CONSENSUS_A93339CE = "未找到 topic [{}] 的配置，无法设置 consensus queue"; public static final String PIPE_LOG_NO_LOCAL_IOTCONSENSUS_DATA_REGION_FOUND_FOR_TOPIC_IN_CONSUMER_6FD0600E = diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java index baef673ca86..65d4b3f102d 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/plan/execution/config/executor/ClusterConfigTaskExecutor.java @@ -3172,7 +3172,7 @@ public class ClusterConfigTaskExecutor implements IConfigTaskExecutor { // Validate topic config final TopicMeta temporaryTopicMeta = new TopicMeta(topicName, System.currentTimeMillis(), topicAttributes); - if (!temporaryTopicMeta.getConfig().isConsensusMode()) { + if (!temporaryTopicMeta.getConfig().isIncrementalMode()) { try { PipeDataNodeAgent.plugin() .validate( 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 63ebe5b2814..6a482641851 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 @@ -421,9 +421,8 @@ public class SubscriptionBrokerAgent { final String runtimeConsensusImplementation = Objects.nonNull(dataRegionConsensus) ? dataRegionConsensus.getClass().getName() : "null"; return String.format( - "Subscription: cannot %s consensus-based topic(s) %s in consumer group [%s] because " - + "mode=consensus only supports data_region_consensus_protocol_class=%s, but current " - + "configured value is %s (runtime consensus implementation: %s)", + DataNodePipeMessages + .EXCEPTION_SUBSCRIPTION_CANNOT_ARG_CONSENSUS_BASED_TOPIC_S_ARG_IN_CONSUMER_GROUP_ARG_BECAUSE_MODE_INCREMENTAL_ONLY_SUPPORTS_DATA_REGION_CONSENSUS_PROTOCOL_CLASS_ARG_BUT_CURRENT_CONFIGURED_VALUE_IS_ARG_RUNTIME_CONSENSUS_IMPLEMENTATION_ARG_6F21ED67, operation, topicNames, consumerGroupId, diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java index c307f281e92..0e1ea95a314 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/subscription/broker/consensus/ConsensusSubscriptionSetupHandler.java @@ -58,9 +58,9 @@ import java.util.concurrent.ConcurrentHashMap; /** * Handles setup and teardown of consensus-based subscription queues on DataNode. * - * <p>For each consensus-mode topic subscribed by a consumer group, this handler discovers matching - * local IoTConsensus DataRegions, builds the appropriate log-to-tablet converter, and binds one - * queue per region to the consensus subscription broker. + * <p>For each incremental-mode topic subscribed by a consumer group, this handler discovers + * matching local IoTConsensus DataRegions, builds the appropriate log-to-tablet converter, and + * binds one queue per region to the consensus subscription broker. */ public class ConsensusSubscriptionSetupHandler { @@ -287,7 +287,7 @@ public class ConsensusSubscriptionSetupHandler { public static boolean isConsensusBasedTopic(final String topicName) { try { final String topicMode = SubscriptionAgent.topic().getTopicMode(topicName); - final boolean result = TopicConstant.MODE_CONSENSUS_VALUE.equalsIgnoreCase(topicMode); + final boolean result = TopicConstant.MODE_INCREMENTAL_VALUE.equalsIgnoreCase(topicMode); LOGGER.debug( DataNodePipeMessages.PIPE_LOG_ISCONSENSUSBASEDTOPIC_CHECK_FOR_TOPIC_MODE_RESULT_19EFA0F9, topicName, @@ -304,6 +304,18 @@ public class ConsensusSubscriptionSetupHandler { } } + private static boolean isConsensusBasedTopicRequired(final String topicName) { + final String topicMode = SubscriptionAgent.topic().getTopicMode(topicName); + if (Objects.isNull(topicMode)) { + throw new SubscriptionException( + String.format( + DataNodePipeMessages + .EXCEPTION_TOPIC_METADATA_FOR_ARG_IS_UNAVAILABLE_DURING_CONSENSUS_SUBSCRIPTION_SETUP_A1949F20, + topicName)); + } + return TopicConstant.MODE_INCREMENTAL_VALUE.equalsIgnoreCase(topicMode); + } + public static void setupConsensusSubscriptions( final String consumerGroupId, final Set<String> topicNames) { final IConsensus dataRegionConsensus = DataRegionConsensusImpl.getInstance(); @@ -313,7 +325,7 @@ public class ConsensusSubscriptionSetupHandler { Objects.nonNull(dataRegionConsensus) ? dataRegionConsensus.getClass().getName() : "null"; LOGGER.warn( DataNodePipeMessages - .PIPE_LOG_SKIPPING_SETUP_OF_CONSENSUS_BASED_SUBSCRIPTIONS_FOR_CONSUMER_A7B2C812, + .PIPE_LOG_SKIPPING_SETUP_OF_CONSENSUS_BASED_SUBSCRIPTIONS_FOR_CONSUMER_46BEE6E4, consumerGroupId, ConsensusFactory.IOT_CONSENSUS, configuredProtocol,
