Data Subscription API

About 6 min

Data Subscription API

The IoTDB table-model data subscription API allows applications to manage Topics through the Java SDK, actively pull subscription data, and commit consumption progress. For functional concepts and SQL syntax, see Data Subscription.

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

1. Core Steps

Create a Topic with ISubscriptionTableSession, defining the subscribed data by database, table, and time range.
Subscribe to the Topic. A Consumer can subscribe only to an existing Topic; Consumers in the same Consumer Group that subscribe to the same Topic share the workload.
Consume data by actively calling poll() to obtain SubscriptionMessage objects.
Commit progress automatically, or disable automatic commits and call commitSync() or commitAsync().
Call unsubscribe(). When a Consumer closes, it exits the Consumer Group.

For a Topic with mode=consensus:

Consensus subscription is available only for DataRegions that use IoTConsensus.
start-time and end-time cannot be used to specify a consumption range.
The consumption start point is established when the Consumer Group first subscribes to the Topic successfully. Data written before the subscription is not replayed.
Only SubscriptionRecordHandler is supported; TsFile format is not supported.

Even when different clients use the same Consumer Group ID and Consumer ID, the server treats them as separate connections and distributes the workload between them.

2. Detailed Steps

This section demonstrates the core development flow and does not include every parameter and interface. For the complete reference, see Full API Reference.

2.1 Create a Maven Project

Create a Maven project and add the following dependency (JDK >= 17, Maven >= 3.6):

<dependencies>
    <dependency>
      <groupId>org.apache.iotdb</groupId>
      <artifactId>iotdb-session</artifactId>
      <!-- The version number is the same as the database version. -->
      <version>${project.version}</version>
    </dependency>
</dependencies>

2.2 Examples

2.2.1 Manage a Regular Topic

import java.util.Properties;
import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.apache.iotdb.session.subscription.ISubscriptionTableSession;
import org.apache.iotdb.session.subscription.SubscriptionTableSessionBuilder;

public class TopicOperationExample {

  public static void main(String[] args) throws Exception {
    try (final ISubscriptionTableSession session =
        new SubscriptionTableSessionBuilder()
            .host("127.0.0.1")
            .port(6667)
            .username("root")
            .password("TimechoDB@2021")
            .build()) {

      final Properties config = new Properties();
      config.put(TopicConstant.DATABASE_KEY, "db.*");
      config.put(TopicConstant.TABLE_KEY, "test.*");
      config.put(TopicConstant.START_TIME_KEY, 25);
      config.put(TopicConstant.END_TIME_KEY, 75);
      config.put(TopicConstant.STRICT_KEY, "true");
      config.put(
          TopicConstant.FORMAT_KEY,
          TopicConstant.FORMAT_RECORD_HANDLER_VALUE);

      session.createTopicIfNotExists("topic1", config);
      session.getTopic("topic1").ifPresent(System.out::println);
      session.getSubscriptions("topic1").forEach(System.out::println);
    }
  }
}

2.2.2 Create a Consensus Subscription Topic

The following example creates a consensus subscription Topic with a Topic configuration string:

import java.util.Properties;
import org.apache.iotdb.session.subscription.ISubscriptionTableSession;
import org.apache.iotdb.session.subscription.SubscriptionTableSessionBuilder;

public class ConsensusTopicOperationExample {

  public static void main(String[] args) throws Exception {
    try (final ISubscriptionTableSession session =
        new SubscriptionTableSessionBuilder()
            .host("127.0.0.1")
            .port(6667)
            .username("root")
            .password("TimechoDB@2021")
            .build()) {

      final Properties config = new Properties();
      config.setProperty("mode", "consensus");
      config.setProperty("database", "factory_db");
      config.setProperty("table", "sensor_data");
      config.setProperty("column", ".*");
      config.setProperty("format", "SubscriptionRecordHandler");
      config.setProperty("retention.bytes", "536870912");
      config.setProperty("retention.ms", "3600000");

      session.createTopicIfNotExists("consensus_topic", config);
      session.getTopic("consensus_topic").ifPresent(System.out::println);
    }
  }
}

Do not set unsupported settings such as start-time, end-time, strict, or order-mode when creating a consensus subscription Topic.

2.2.3 Consume Data by Row

Row data is obtained through SubscriptionMessage.getResultSets().

import java.util.List;
import org.apache.iotdb.session.subscription.consumer.ISubscriptionTablePullConsumer;
import org.apache.iotdb.session.subscription.consumer.table.SubscriptionTablePullConsumerBuilder;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.tsfile.read.query.dataset.ResultSet;

public class RecordSubscriptionExample {

  public static void main(String[] args) throws Exception {
    try (final ISubscriptionTablePullConsumer consumer =
        new SubscriptionTablePullConsumerBuilder()
            .host("127.0.0.1")
            .port(6667)
            .username("root")
            .password("TimechoDB@2021")
            .consumerId("c1")
            .consumerGroupId("cg1")
            .build()) {

      consumer.open();
      consumer.subscribe("topic1");

      while (true) {
        final List<SubscriptionMessage> messages = consumer.poll(10_000L);
        for (final SubscriptionMessage message : messages) {
          for (final ResultSet resultSet : message.getResultSets()) {
            final SubscriptionRecordHandler.SubscriptionResultSet recordSet =
                (SubscriptionRecordHandler.SubscriptionResultSet) resultSet;
            System.out.println(recordSet.getDatabaseName());
            System.out.println(recordSet.getTableName());
            System.out.println(recordSet.getColumnNames());
            System.out.println(recordSet.getColumnTypes());
            System.out.println(recordSet.getColumnCategories());
            while (recordSet.hasNext()) {
              System.out.println(recordSet.nextRecord());
            }
          }
        }
        // autoCommit is true by default.
      }
    }
  }
}

This consumer code also applies to consensus subscription Topics. Messages from a consensus subscription contain only data that entered the consensus path after the Consumer Group established the subscription relationship.

2.2.4 Commit Consumption Progress Manually

Disable automatic commits when progress should be committed only after business processing succeeds:

import java.util.List;
import org.apache.iotdb.session.subscription.consumer.ISubscriptionTablePullConsumer;
import org.apache.iotdb.session.subscription.consumer.table.SubscriptionTablePullConsumerBuilder;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
import org.apache.iotdb.session.subscription.payload.SubscriptionRecordHandler;
import org.apache.tsfile.read.query.dataset.ResultSet;

public class ManualCommitSubscriptionExample {

  public static void main(String[] args) throws Exception {
    try (final ISubscriptionTablePullConsumer consumer =
        new SubscriptionTablePullConsumerBuilder()
            .host("127.0.0.1")
            .port(6667)
            .username("root")
            .password("TimechoDB@2021")
            .consumerId("c1")
            .consumerGroupId("cg1")
            .autoCommit(false)
            .build()) {

      consumer.open();
      consumer.subscribe("consensus_topic");

      while (true) {
        final List<SubscriptionMessage> messages = consumer.poll(10_000L);
        if (messages.isEmpty()) {
          continue;
        }

        for (final SubscriptionMessage message : messages) {
          for (final ResultSet resultSet : message.getResultSets()) {
            final SubscriptionRecordHandler.SubscriptionResultSet recordSet =
                (SubscriptionRecordHandler.SubscriptionResultSet) resultSet;
            while (recordSet.hasNext()) {
              // Downstream processing should be idempotent.
              System.out.println(recordSet.nextRecord());
            }
          }
        }

        consumer.commitSync(messages);
      }
    }
  }
}

If the Consumer exits after processing but before the commit succeeds, unacknowledged messages may be delivered again after restart or reassignment. Data subscription therefore provides at-least-once semantics, not exactly-once semantics.

When the client processor still buffers messages, call drainBufferedMessages() before closing, then process and commit the returned messages.

This scenario applies only to regular subscriptions. It does not apply to Topics with mode=consensus.

Create a TsFile-format Topic:

CREATE TOPIC topic_tsfile
WITH (
  'database' = 'database1',
  'table' = 'table1',
  'format' = 'SubscriptionTsFileHandler'
);

Consume the TsFile:

import org.apache.iotdb.session.subscription.consumer.ISubscriptionTablePullConsumer;
import org.apache.iotdb.session.subscription.consumer.table.SubscriptionTablePullConsumerBuilder;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
import org.apache.iotdb.session.subscription.payload.SubscriptionTsFileHandler;
import org.apache.tsfile.read.v4.ITsFileReader;

public class TsFileSubscriptionExample {

  public static void main(String[] args) throws Exception {
    try (final ISubscriptionTablePullConsumer consumer =
        new SubscriptionTablePullConsumerBuilder()
            .host("127.0.0.1")
            .port(6667)
            .username("root")
            .password("TimechoDB@2021")
            .consumerId("c1")
            .consumerGroupId("cg1")
            .build()) {

      consumer.open();
      consumer.subscribe("topic_tsfile");

      while (true) {
        for (final SubscriptionMessage message : consumer.poll(10_000L)) {
          final SubscriptionTsFileHandler handler = message.getTsFile();
          try (final ITsFileReader reader = handler.openTableReader()) {
            // Read the subscribed table-model TsFile.
          }
        }
      }
    }
  }
}

3. Parameters and Interfaces

3.1 Common Parameters

3.1.1 Common Consumer Configuration

Parameter	Default	Description
`host`	`127.0.0.1`	DataNode RPC host.
`port`	`6667`	DataNode RPC port.
`nodeUrls`	None	DataNode RPC endpoints.
`username`	`root`	Username.
`password`	`root`	Password.
`encryptedPassword`	None	Encrypted password.
`consumerId`	Automatically assigned	Globally unique Consumer ID assigned when the Consumer opens.
`consumerGroupId`	Automatically assigned	Globally unique Consumer Group ID assigned when the Consumer opens.
`heartbeatIntervalMs`	`30000`, minimum `1000`	Heartbeat interval in milliseconds.
`endpointsSyncIntervalMs`	`120000`, minimum `5000`	Cluster endpoint synchronization interval in milliseconds.
`fileSaveDir`	`<user.dir>/iotdb-subscription`	Temporary TsFile directory.
`fileSaveFsync`	`false`	Whether to run fsync when saving a TsFile.
`connectionTimeoutInMs`	`0`	Connection timeout.
`maxPollParallelism`	`1`	Maximum parallel poll count.

3.1.2 Pull Consumer Configuration

Parameter	Default	Description
`autoCommit`	`true`	Whether to commit consumption progress automatically. When `false`, a commit API must be called manually.
`autoCommitIntervalMs`	`5000`, minimum `500`	Automatic commit interval. Applies only when `autoCommit=true`.

3.2 Interfaces

3.2.1 ISubscriptionTableSession

Manages table-model Topics and queries Topics and subscription relationships.

Method	Description	Return value	Main exception
`open()`	Opens the Session and connects to IoTDB.	`void`	`IoTDBConnectionException`
`createTopic(String topicName)`	Creates a Topic with the default configuration.	`void`	`IoTDBConnectionException`, `StatementExecutionException`
`createTopicIfNotExists(String topicName)`	Creates a Topic when it does not exist.	`void`	Same as above
`createTopic(String topicName, Properties properties)`	Creates a Topic with properties.	`void`	Same as above
`createTopicIfNotExists(String topicName, Properties properties)`	Creates a Topic with properties when it does not exist.	`void`	Same as above
`alterTopic(String topicName, Properties properties)`	Modifies mutable Topic settings.	`void`	Same as above
`alterTopicOwner(String topicName, String ownerId, long ownerEpoch)`	Modifies Topic Owner information.	`void`	Same as above
`alterTopicOwner(String topicName, String ownerId, long ownerEpoch, Long maxOwnerEpoch)`	Modifies Topic Owner and maximum Epoch.	`void`	Same as above
`dropTopic(String topicName)`	Drops a Topic.	`void`	Same as above
`dropTopicIfExists(String topicName)`	Drops a Topic when it exists.	`void`	Same as above
`dropSubscription(String subscriptionId)`	Drops a subscription relationship.	`void`	Same as above
`dropSubscriptionIfExists(String subscriptionId)`	Drops a subscription relationship when it exists.	`void`	Same as above
`getTopics()`	Gets all Topics.	`Set<Topic>`	Same as above
`getTopic(String topicName)`	Gets a Topic, or an empty value when it does not exist.	`Optional<Topic>`	Same as above
`getSubscriptions()`	Gets all subscription relationships.	`Set<Subscription>`	Same as above
`getSubscriptions(String topicName)`	Gets subscription relationships for a Topic.	`Set<Subscription>`	Same as above
`close()`	Closes the Session.	`void`	`Exception`

3.2.2 ISubscriptionTablePullConsumer

Subscribes to Topics, actively pulls messages, and commits consumption progress.

Method	Description	Return value	Main exception
`open()`	Opens the Consumer.	`void`	`SubscriptionException`
`subscribe(String topicName)`	Subscribes to one Topic.	`void`	`SubscriptionException`
`subscribe(String... topicNames)`	Subscribes to multiple Topics.	`void`	`SubscriptionException`
`subscribe(Set<String> topicNames)`	Subscribes to multiple Topics.	`void`	`SubscriptionException`
`unsubscribe(String topicName)`	Unsubscribes from one Topic.	`void`	`SubscriptionException`
`unsubscribe(String... topicNames)`	Unsubscribes from multiple Topics.	`void`	`SubscriptionException`
`unsubscribe(Set<String> topicNames)`	Unsubscribes from multiple Topics.	`void`	`SubscriptionException`
`poll(Duration timeout)`	Pulls messages. The timeout is the maximum wait when no message is available and does not limit the result count.	`List<SubscriptionMessage>`	`SubscriptionException`
`poll(long timeoutMs)`	Pulls messages with a maximum wait in milliseconds.	`List<SubscriptionMessage>`	`SubscriptionException`
`poll(Set<String> topicNames, Duration timeout)`	Pulls messages from the specified Topics.	`List<SubscriptionMessage>`	`SubscriptionException`
`poll(Set<String> topicNames, long timeoutMs)`	Pulls messages from the specified Topics.	`List<SubscriptionMessage>`	None
`drainBufferedMessages()`	Drains messages buffered by the client processor.	`List<SubscriptionMessage>`	`SubscriptionException`
`commitSync(SubscriptionMessage message)`	Commits one message synchronously.	`void`	`SubscriptionException`
`commitSync(Iterable<SubscriptionMessage> messages)`	Commits multiple messages synchronously.	`void`	`SubscriptionException`
`commitAsync(SubscriptionMessage message)`	Commits one message asynchronously.	`CompletableFuture<Void>`	None
`commitAsync(Iterable<SubscriptionMessage> messages)`	Commits multiple messages asynchronously.	`CompletableFuture<Void>`	None
`commitAsync(SubscriptionMessage message, AsyncCommitCallback callback)`	Commits one message asynchronously and invokes a callback.	`void`	None
`commitAsync(Iterable<SubscriptionMessage> messages, AsyncCommitCallback callback)`	Commits multiple messages asynchronously and invokes a callback.	`void`	None
`seekToBeginning(String topicName)`	Moves the consumption position to the beginning.	`void`	`SubscriptionException`
`seekToEnd(String topicName)`	Moves the consumption position to the end.	`void`	`SubscriptionException`
`positions(String topicName)`	Gets the current position.	`TopicProgress`	`SubscriptionException`
`committedPositions(String topicName)`	Gets the committed position.	`TopicProgress`	`SubscriptionException`
`seek(String topicName, TopicProgress topicProgress)`	Moves to the specified position.	`void`	`SubscriptionException`
`seekAfter(String topicName, TopicProgress topicProgress)`	Moves to the position after the specified progress.	`void`	`SubscriptionException`
`getConsumerId()`	Gets the Consumer ID.	`String`	None
`getConsumerGroupId()`	Gets the Consumer Group ID.	`String`	None
`allTopicMessagesHaveBeenConsumed()`	Tests whether all Topic messages have been consumed.	`boolean`	None
`close()`	Closes the Consumer and exits the Consumer Group.	`void`	`Exception`

3.2.3 SubscriptionMessage

SubscriptionMessage is the basic message unit returned by poll(). The data format is determined by the Topic format: use getResultSets() for row format and getTsFile() for TsFile format.

Method	Description	Return value
`getMessageType()`	Gets the message type.	`short`
`isTimeSelected()`	Tests whether the message was filtered by a time range.	`boolean`
`getResultSets()`	Gets row-format result sets.	`List<ResultSet>`
`getRecordTabletIterator()`	Reads row data through a Tablet iterator.	`Iterator<Tablet>`
`getTsFile()`	Gets the TsFile handler.	`SubscriptionTsFileHandler`
`getWatermarkTimestamp()`	Gets the timestamp carried by a Watermark message.	`long`
`estimateSize()`	Estimates the heap bytes occupied by the message.	`long`
`removeUserData()`	Releases user data in the message.	`void`

Calling a reader that is incompatible with the message format throws SubscriptionIncompatibleHandlerException. Reading a message after its data has been released may throw SubscriptionRuntimeException.

3.2.4 SubscriptionRecordHandler.SubscriptionResultSet

The table-model implementation of a row-data result set.

Method	Description	Return value
`getDatabaseName()`	Gets the database name.	`String`
`getTableName()`	Gets the table name.	`String`
`getColumnNames()`	Gets column names.	Column-name list
`getColumnTypes()`	Gets column data types.	Column-type list
`getColumnCategories()`	Gets column categories.	Column-category list
`hasNext()`	Tests whether another row is available.	`boolean`
`nextRecord()`	Reads the next row.	`RowRecord`

3.2.5 SubscriptionTsFileHandler

Handles TsFiles delivered to the client by regular subscriptions. Consensus subscriptions do not support this handler.

Method	Description	Return value
`getDatabaseName()`	Gets the database containing the TsFile.	`String`
`openTableReader()`	Opens a table-model TsFile Reader.	`ITsFileReader`

When the file is not a table-model TsFile, openTableReader() throws SubscriptionIncompatibleHandlerException. Reading the file may also throw IOException.