Skip to content

Subscription: Rename topic modes to initial and incremental #18370

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
jt2594838 merged 1 commit into from
Aug 3, 2026
Merged

Subscription: Rename topic modes to initial and incremental #18370

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion example/subscription/src/main/java/org/apache/iotdb/ConsensusSubscriptionSessionExample.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -109,7 +109,7 @@ private static void createConsensusTopic(final String topicName, final String pa
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);
Expand Down
2 changes: 1 addition & 1 deletion ...ription/src/main/java/org/apache/iotdb/ConsensusTableModelSubscriptionSessionExample.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -107,7 +107,7 @@ private static void createConsensusTopic(
.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);
Expand Down
2 changes: 1 addition & 1 deletion ...java/org/apache/iotdb/subscription/it/consensus/local/ConsensusSubscriptionITSupport.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -94,7 +94,7 @@ static void createConsensusTopic(final String topicName, final String path) thro
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);
Expand Down
2 changes: 1 addition & 1 deletion ...iotdb/subscription/it/consensus/local/tablemodel/ConsensusSubscriptionTableITSupport.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -134,7 +134,7 @@ static void createConsensusTopic(
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);
Expand Down
2 changes: 1 addition & 1 deletion ...iption/it/consensus/local/tablemodel/IoTDBConsensusSubscriptionColumnFilterClusterIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -150,7 +150,7 @@ private static void createOwnedConsensusTopic(
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);
Expand Down
5 changes: 3 additions & 2 deletions ...ava/org/apache/iotdb/subscription/it/dual/tablemodel/IoTDBSubscriptionColumnFilterIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1015,7 +1015,7 @@ private void createTopic(
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(
Expand All @@ -1032,7 +1032,8 @@ private void createTopic(
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(
Expand Down
2 changes: 1 addition & 1 deletion ...java/org/apache/iotdb/subscription/it/local/tablemodel/IoTDBSubscriptionPermissionIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -390,7 +390,7 @@ public void testColumnFilterTopicRequiresSystemPermission() throws Exception {
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);
Expand Down
2 changes: 1 addition & 1 deletion ...triple/treemodel/regression/pullconsumer/loose_range/IoTDBAllTsDatasetPullConsumerIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -75,7 +75,7 @@ public void setUp() throws Exception {
"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);
Expand Down
2 changes: 1 addition & 1 deletion .../triple/treemodel/regression/pullconsumer/loose_range/IoTDBAllTsTsfilePullConsumerIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -78,7 +78,7 @@ public void setUp() throws Exception {
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);
Expand Down
2 changes: 1 addition & 1 deletion ...e/treemodel/regression/pullconsumer/loose_range/IoTDBPathDeviceDataSetPullConsumerIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,7 @@ public void setUp() throws Exception {
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);
Expand Down
2 changes: 1 addition & 1 deletion ...le/treemodel/regression/pullconsumer/loose_range/IoTDBPathDeviceTsfilePullConsumerIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,7 @@ public void setUp() throws Exception {
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);
Expand Down
2 changes: 1 addition & 1 deletion ...riple/treemodel/regression/pullconsumer/loose_range/IoTDBTimeTsDatasetPullConsumerIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,7 @@ public void setUp() throws Exception {
"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);
Expand Down
2 changes: 1 addition & 1 deletion ...triple/treemodel/regression/pullconsumer/loose_range/IoTDBTimeTsTsfilePullConsumerIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,7 @@ public void setUp() throws Exception {
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);
Expand Down
4 changes: 2 additions & 2 deletions ...e/treemodel/regression/pushconsumer/loose_range/IoTDBLooseAllTsDatasetPushConsumerIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@
* DataSet
* pattern: ts
* loose-range: all
* mode: live
* mode: initial
*/
@RunWith(IoTDBTestRunner.class)
@Category({MultiClusterIT2SubscriptionTreeRegressionConsumer.class})
Expand All @@ -82,7 +82,7 @@ public void setUp() throws Exception {
"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);
Expand Down
4 changes: 2 additions & 2 deletions ...iple/treemodel/regression/pushconsumer/loose_range/IoTDBLooseAllTsfilePushConsumerIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,7 +58,7 @@

/***
* push consumer
* mode: live
* mode: initial
* pattern: db
* loose-range: all
*/
Expand All @@ -84,7 +84,7 @@ public void setUp() throws Exception {
"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);
Expand Down
2 changes: 1 addition & 1 deletion ...eemodel/regression/pushconsumer/loose_range/IoTDBPathLooseDeviceTsfilePushConsumerIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,7 @@ public void setUp() throws Exception {
"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);
Expand Down
2 changes: 1 addition & 1 deletion .../treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseTsDatasetPushConsumerIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,7 @@ public void setUp() throws Exception {
"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);
Expand Down
2 changes: 1 addition & 1 deletion ...ple/treemodel/regression/pushconsumer/loose_range/IoTDBPathLooseTsfilePushConsumerIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,7 @@ public void setUp() throws Exception {
"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);
Expand Down
2 changes: 1 addition & 1 deletion .../treemodel/regression/pushconsumer/loose_range/IoTDBPathTsLooseDatasetPushConsumerIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,7 +81,7 @@ public void setUp() throws Exception {
"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);
Expand Down
4 changes: 2 additions & 2 deletions .../treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsDatasetPushConsumerIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@
* DataSet
* pattern: ts
* loose-range: time
* live
* initial
*/
@RunWith(IoTDBTestRunner.class)
@Category({MultiClusterIT2SubscriptionTreeRegressionConsumer.class})
Expand All @@ -84,7 +84,7 @@ public void setUp() throws Exception {
"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);
Expand Down
4 changes: 2 additions & 2 deletions ...e/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsTsfilePushConsumerIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@
import static org.apache.iotdb.subscription.it.IoTDBSubscriptionITConstant.AWAIT;

/***
* mode: live
* mode: initial
* loose-range:path
* format: tsfile
*/
Expand All @@ -83,7 +83,7 @@ public void setUp() throws Exception {
"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);
Expand Down
2 changes: 1 addition & 1 deletion ...ple/treemodel/regression/pushconsumer/loose_range/IoTDBTimeLooseTsfilePushConsumerIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,7 +77,7 @@ public void setUp() throws Exception {
"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);
Expand Down
4 changes: 2 additions & 2 deletions .../treemodel/regression/pushconsumer/loose_range/IoTDBTimeTsLooseDatasetPushConsumerIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,7 +56,7 @@
* DataSet
* pattern: ts
* time loose
* live
* initial
*/
@RunWith(IoTDBTestRunner.class)
@Category({MultiClusterIT2SubscriptionTreeRegressionConsumer.class})
Expand All @@ -83,7 +83,7 @@ public void setUp() throws Exception {
"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);
Expand Down
2 changes: 1 addition & 1 deletion ...ription/src/main/i18n/en/org/apache/iotdb/rpc/subscription/i18n/SubscriptionMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -270,7 +270,7 @@ private 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 = ", ";
Expand Down
2 changes: 1 addition & 1 deletion ...ription/src/main/i18n/zh/org/apache/iotdb/rpc/subscription/i18n/SubscriptionMessages.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -219,7 +219,7 @@ private 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 = "，";
Expand Down
51 changes: 41 additions & 10 deletions ...ient/subscription/src/main/java/org/apache/iotdb/rpc/subscription/config/TopicConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,7 @@
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;
Expand All @@ -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);
Expand Down Expand Up @@ -84,8 +85,10 @@ public TopicConfig(final Map<String, String> attributes) {

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");
Expand Down Expand Up @@ -125,20 +128,48 @@ public boolean isSnapshotMode() {
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) {
return MODE_VALUE_SET.contains(normalizeMode(mode));
}

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() {
Expand Down Expand Up @@ -223,12 +254,12 @@ public Map<String, String> getAttributesWithSourceRealtimeMode() {
}

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() {
Expand Down
Loading
Loading