Data Subscription

1. Feature Overview

The IoTDB data subscription module, referred to as the IoTDB subscription client, provides a streaming data consumption method that differs from data queries. It follows the basic concepts and logic of message queue systems such as Kafka and provides data subscription and consumption APIs. It is not intended to completely replace message queues, but to simplify scenarios that require lightweight streaming data acquisition.

The IoTDB subscription client is especially useful in the following scenarios:

1. Obtain newly written data in real time: Continuously pull newly written data without frequently running scheduled queries, reducing application complexity and query load. This is suitable for dashboards, supervisory control, and similar scenarios that require timely updates.
2. Integrate with third-party systems: Downstream systems such as Flink, Kafka, DataX, and MySQL can actively pull data as subscription clients, without requiring a separate push component in IoTDB for every system.
3. Incrementally back up TsFiles: Subscribe to newly generated TsFiles and archive incremental data to a specified storage location for periodic backups or off-site storage.
4. Run low-latency real-time computation: Consensus subscription consumes data directly from the IoTConsensus write path, reducing dependency on Pipe extraction and historical file scans.

Note: This feature is supported starting from V2.0.12.1.

2. Key Concepts

The IoTDB subscription client has three core concepts: Topic, Consumer, and Consumer Group.

2.1 Topic

A Topic is a data space in IoTDB that can be subscribed to. In the table model, the data range is determined by the following settings:

• database: Database name or matching expression.
• table: Table name or matching expression.
• column: Column name or matching expression.

A regular subscription can also use start-time and end-time to limit the Event Time range. Unlike Kafka, IoTDB allows a Topic to be created after data has already been written.

Topics support the following modes:

Mode	Description
snapshot	A static data set. A snapshot is generated at the time a Consumer Group subscribes to the Topic and does not continuously include data written after that snapshot.
incremental	Consensus subscription. It consumes real-time incremental data from the IoTConsensus write log. Only data entering the consensus path after the Consumer Group first subscribes successfully is consumed; historical data is not replayed.

The table model supports the following output formats:

Format	Description
SubscriptionRecordHandler	Consumes data by row.
SubscriptionTsFileHandler	Consumes data by TsFile.

Consensus subscription supports only SubscriptionRecordHandler, not SubscriptionTsFileHandler.

2.2 Consumer

A Consumer is a subscription client that subscribes to Topics, receives data, and commits consumption progress.

The table model currently provides a Pull Consumer. Application code actively calls poll() to pull data and can choose automatic or manual progress commits.

2.3 Consumer Group

Consumers with the same Consumer Group ID belong to the same Consumer Group:

• A Consumer Group can contain multiple Consumers, while a Consumer can join only one Consumer Group.
• Consumers in the same Consumer Group that subscribe to the same Topic share the consumption workload. A message is assigned to only one Consumer in the group.
• Different Consumer Groups are independent and can consume the same Topic separately.
• Messages that have not been committed successfully may be delivered again after a Consumer restarts or messages are reassigned. Data subscription therefore provides at-least-once semantics, not exactly-once semantics.

2.4 Considerations

• Real-time data is consumed in arrival order within a single Region. Global ordering across Regions and ordering by Event Time are not guaranteed.
• Uncommitted messages may be delivered again after a Consumer restart or reassignment.
• Downstream systems should use idempotent writes or deduplication to handle duplicate data. Applications that depend on Event Time ordering must handle out-of-order data themselves.
• The table model uses its own namespace for Session, Consumer, Topic, and Subscription resources. Topics capture only table-model data and are visible only in this namespace.

3. SQL Statements

3.1 Topic Management

IoTDB supports creating, dropping, and showing Topics through SQL statements. The Topic lifecycle is shown below:

3.1.1 Create a Topic

CREATE TOPIC [IF NOT EXISTS] <topicName>
WITH (
  [<parameter> = <value>,]
);

IF NOT EXISTS prevents an error when the Topic already exists.

3.1.1.1 Regular Subscription

Regular subscriptions support the snapshot mode and can filter data by database, table, column, and time range.

Parameter	Default	Description
database	.*	Database to subscribe to. Regular expressions are supported.
table	.*	Table to subscribe to. Regular expressions are supported.
column	All columns	Columns to subscribe to. Regular expressions are supported.
start-time	MIN_VALUE	Start of the Event Time range. Supports ISO timestamps, long values matching the database timestamp precision, and the special value now.
end-time	MAX_VALUE	End of the Event Time range. The supported formats are the same as for start-time.
mode	snapshot	Exports only the current snapshot.
format	SubscriptionRecordHandler	Supports SubscriptionRecordHandler and SubscriptionTsFileHandler.
order-mode	leader-only	Supports leader-only, multi-writer, and per-writer.
loose-range	""	Controls coarse filtering of the data and time ranges.
strict	true	Whether to filter data strictly according to the Topic range.
processor	do-nothing-processor	Processing plugin applied to the original subscription data.

Time-range behavior:

• start-time=MIN_VALUE and end-time=now: subscribe only to historical data.
• start-time=now and end-time=MAX_VALUE: subscribe only to real-time data.
• now represents the Topic creation time.

Subscribe to all data:

CREATE TOPIC data_all;

Subscribe by database, table, and time range:

CREATE TOPIC IF NOT EXISTS topic_table1
WITH (
  'database' = 'database1',
  'table' = 'table1',
  'start-time' = '2024-11-01',
  'end-time' = '2024-11-30',
  'strict' = 'false'
);

3.1.1.2 Consensus Subscription

Set mode=incremental when creating a Topic to enable consensus subscription.

• Consensus subscription is available only for DataRegions that use IoTConsensus.
• Consensus subscription targets only real-time incremental data. The consumption start point is established when the Consumer Group first subscribes successfully.
• Data written before the subscription is not replayed, and historical data cannot be recovered through a time range.

Consensus subscription supports the following settings:

Parameter	Default	Description
database	.*	Database matching range.
table	.*	Table matching range.
column	All columns	Column matching range. Regular expressions are supported.
format	SubscriptionRecordHandler	Only SubscriptionRecordHandler is supported.
retention.bytes	536870912	Maximum retained WAL size in bytes. Supports a positive long value or -1; -1 means unlimited.
retention.ms	-1	Maximum WAL retention time in milliseconds. Supports a positive long value or -1; -1 means unlimited.

retention.bytes and retention.ms cannot be 0, less than -1, or a non-long value. When both limits are set, old WAL entries may be cleaned when either limit is reached. Cleaned WAL entries cannot be used for recovery or replay. These settings cannot be modified after the Topic is created.

Example:

CREATE TOPIC IF NOT EXISTS consensus_topic
WITH (
  'mode' = 'incremental',
  'database' = 'factory_db',
  'table' = 'sensor_data',
  'format' = 'SubscriptionRecordHandler',
  'column' = '.*',
  'retention.bytes' = '536870912',
  'retention.ms' = '3600000'
);

3.1.2 Drop a Topic

Only an unsubscribed Topic can be dropped. When a Topic is dropped, its associated consumption progress is cleared.

DROP TOPIC [IF EXISTS] <topicName>;

IF EXISTS runs the operation only when the Topic exists and prevents an error for a nonexistent Topic.

3.1.3 Show Topics

SHOW TOPICS;
SHOW TOPIC <topicName>;

Result set:

[TopicName|TopicConfigs]

• TopicName: Topic ID.
• TopicConfigs: Topic configuration, currently containing the parameters specified in the WITH clause when the Topic was created.

3.1.4 View Topic Creation Information

Use this statement to obtain the creation statement for a specified Topic. This feature is supported since V2.0.12.1.

Syntax:

SHOW CREATE TOPIC <topicName>;

Result:

Column Name	Type	Description
Topic	TEXT	Topic name.
Create Topic	TEXT	A CREATE TOPIC statement that can be used to recreate the static Topic definition.

Notes:

1. Only table-model-visible Topics are supported, and data subscription must be enabled. When the feature is disabled, a related not-enabled error is returned.
2. If the Topic does not exist or is not visible to the current model, a Topic-not-found or permission-related error is returned.
3. The Topic name is always enclosed in double quotation marks. Property names and values are output as string literals, and properties are sorted stably by property name. If there are no user-visible properties, the WITH (...) clause is omitted.
4. Properties with the __system or __audit prefix are omitted. Only the static Topic definition is returned; Consumer Groups, subscription relationships, consumption positions, and consumption progress are not included.

Example:

View the creation information for the existing Topic topic_table1. This table-model Topic subscribes to table1 in database1 and sets a start time, end time, and non-strict filtering:

SHOW CREATE TOPIC topic_table1;

Result:

+------------+---------------------------------------------------------------------------------------------------------------------------+
|       Topic|                                                                                                               Create Topic|
+------------+---------------------------------------------------------------------------------------------------------------------------+
|topic_table1|CREATE TOPIC "topic_table1" WITH ('database'='database1','end-time'='2024-11-30','start-time'='2024-11-01','strict'='false','table'='table1')|
+------------+---------------------------------------------------------------------------------------------------------------------------+

3.2 Show Subscription Status

SHOW SUBSCRIPTIONS;
SHOW SUBSCRIPTIONS ON <topicName>;

Result set:

[SubscriptionID|TopicName|ConsumerGroupName|SubscribedConsumers]

• SubscriptionID: Unique identifier of the subscription relationship. It can be used to manage or drop a specific relationship.
• TopicName: Topic name.
• ConsumerGroupName: Consumer Group ID.
• SubscribedConsumers: IDs of the Consumers that currently subscribe to this Topic.

4. API

In addition to SQL statements, IoTDB provides Java native APIs for data subscription. For details, see Data Subscription API.

4.1 Authorization

• Metadata authorization for SQL statements
  • Creating or dropping a Topic requires the SYSTEM privilege.
  • A user with the SYSTEM privilege can view global resources when querying Topics and subscription relationships. Other users can view only their own resources.
• Runtime authorization through client parameters
  • A Consumer supplies username and password through Properties or a Builder. Identity is verified by open(). Authorization information is passed to the underlying subscription Pipe by subscribe(), and data query privileges are checked dynamically during consumption. Data without the required privilege is skipped or reported according to settings such as if-no-privileges.
• Consumer Group credential consistency
  • All Consumers in the same Consumer Group must use the same username and password. The first Consumer that opens successfully establishes the authorization baseline for the group. Later Consumers can join only when their credentials match.