Data Subscription API

About 6 min

Data Subscription API

The IoTDB tree-model data subscription API allows applications to manage Topics through the Java SDK, obtain data with Pull or Push Consumers, 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: Define the time series to subscribe to by path and optional time range.
Subscribe to a Topic: A Consumer can subscribe only to an existing Topic. Consumers in the same Consumer Group that subscribe to the same Topic share the consumption workload.
Consume data: A Pull Consumer actively calls poll(). A Push Consumer receives callbacks through a user-provided ConsumeListener.
Commit progress: A Pull Consumer can use automatic commits, or disable them and call commitSync() or commitAsync().
Unsubscribe: Call unsubscribe(). Closing a Consumer removes it from the Consumer Group and cancels its existing subscriptions.

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

2.1 Create a Maven Project

Create a Maven project and add iotdb-session. The dependency version should match the database version.

<dependencies>
  <dependency>
    <groupId>org.apache.iotdb</groupId>
    <artifactId>iotdb-session</artifactId>
    <version>${project.version}</version>
  </dependency>
</dependencies>

Runtime requirements:

JDK 17 or later.
Maven 3.6 or later.
Do not use a newer client with an older server.

2.2 Examples

2.2.1 Topic Operations

Use SubscriptionTreeSessionBuilder to create a tree-model subscription Session. build() only constructs the object; call open() before performing Topic operations.

import java.util.Optional;
import java.util.Properties;
import java.util.Set;
import org.apache.iotdb.rpc.subscription.config.TopicConstant;
import org.apache.iotdb.session.subscription.ISubscriptionTreeSession;
import org.apache.iotdb.session.subscription.SubscriptionTreeSessionBuilder;
import org.apache.iotdb.session.subscription.model.Topic;

public class TopicOperationExample {

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

      session.open();

      final Properties topicConfig = new Properties();
      topicConfig.setProperty(TopicConstant.PATH_KEY, "root.**");
      session.createTopicIfNotExists("allData", topicConfig);

      final Set<Topic> topics = session.getTopics();
      System.out.println(topics);

      final Optional<Topic> allData = session.getTopic("allData");
      allData.ifPresent(System.out::println);
    }
  }
}

Before V2.0.6.x, the default password is root. Use the credentials configured for your deployment, and do not hard-code passwords in production code.

2.2.2 Consume Records in Pull Mode

SubscriptionMessage.getRecordTabletIterator() returns a Tablet iterator. Each Tablet contains the device, timestamps, measurement schemas, and values.

import java.util.Iterator;
import java.util.List;
import org.apache.iotdb.session.subscription.consumer.ISubscriptionTreePullConsumer;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePullConsumerBuilder;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessageType;
import org.apache.tsfile.write.record.Tablet;

public class RecordSubscriptionExample {

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

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

      while (true) {
        final List<SubscriptionMessage> messages = consumer.poll(10_000L);
        for (final SubscriptionMessage message : messages) {
          if (message.getMessageType()
              != SubscriptionMessageType.RECORD_HANDLER.getType()) {
            continue;
          }

          final Iterator<Tablet> tablets = message.getRecordTabletIterator();
          while (tablets.hasNext()) {
            final Tablet tablet = tablets.next();
            for (int row = 0; row < tablet.getRowSize(); row++) {
              System.out.printf(
                  "device=%s, time=%d%n",
                  tablet.getDeviceId(), tablet.getTimestamp(row));
              for (int column = 0; column < tablet.getSchemas().size(); column++) {
                System.out.printf(
                    "  %s=%s%n",
                    tablet.getSchemas().get(column).getMeasurementName(),
                    tablet.getValue(row, column));
              }
            }
          }
        }
      }
    }
  }
}

You can also call message.getResultSets() to obtain a List<org.apache.tsfile.read.query.dataset.ResultSet>.

The argument to poll(timeoutMs) is the maximum time to wait when no message is available. It does not limit the number of messages returned by one call.

2.2.3 Commit Consumption Progress Manually

When consumption progress should be committed only after business processing succeeds, disable automatic commits through the Builder:

try (ISubscriptionTreePullConsumer consumer =
    new SubscriptionTreePullConsumerBuilder()
        .host("127.0.0.1")
        .port(6667)
        .username("root")
        .password("TimechoDB@2021")
        .consumerId("c1")
        .consumerGroupId("cg1")
        .autoCommit(false)
        .build()) {

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

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

    for (final SubscriptionMessage message : messages) {
      // Downstream processing should be idempotent.
      System.out.println(message);
    }

    // Commit only after the entire batch has been processed successfully.
    consumer.commitSync(messages);
  }
}

If the Consumer exits after business processing completes 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.

2.2.4 Consume in Push Mode

A Push Consumer configures its callback and acknowledgment strategy through SubscriptionTreePushConsumerBuilder:

AckStrategy.BEFORE_CONSUME: Acknowledge progress before invoking the callback. If the process exits after acknowledgment but before processing, the business logic may not process the message.
AckStrategy.AFTER_CONSUME: Acknowledge progress after the callback succeeds. If the process exits after processing but before acknowledgment, the message may be delivered again.

import org.apache.iotdb.session.subscription.consumer.AckStrategy;
import org.apache.iotdb.session.subscription.consumer.ConsumeResult;
import org.apache.iotdb.session.subscription.consumer.ISubscriptionTreePushConsumer;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePushConsumerBuilder;

public class PushSubscriptionExample {

  public static void main(String[] args) throws Exception {
    try (ISubscriptionTreePushConsumer consumer =
        new SubscriptionTreePushConsumerBuilder()
            .host("127.0.0.1")
            .port(6667)
            .username("root")
            .password("TimechoDB@2021")
            .consumerId("c2")
            .consumerGroupId("cg1")
            .ackStrategy(AckStrategy.AFTER_CONSUME)
            .autoPollIntervalMs(100L)
            .autoPollTimeoutMs(10_000L)
            .consumeListener(
                message -> {
                  try {
                    System.out.println(message);
                    return ConsumeResult.SUCCESS;
                  } catch (Exception e) {
                    return ConsumeResult.FAILURE;
                  }
                })
            .build()) {

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

      // Keep this example running. Production applications should use their lifecycle manager.
      Thread.currentThread().join();
    }
  }
}

The Push Consumer automatically pulls data and invokes the listener according to autoPollIntervalMs and autoPollTimeoutMs.

First create a TsFile-format Topic. The format value is SubscriptionTsFileHandler:

CREATE TOPIC topic_all_tsfile
WITH (
  'path' = 'root.**',
  'format' = 'SubscriptionTsFileHandler'
);

Then use a Pull Consumer to obtain files. getTsFile() returns a SubscriptionTsFileHandler; open a tree-model file with openTreeReader():

import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.List;
import org.apache.iotdb.session.subscription.consumer.ISubscriptionTreePullConsumer;
import org.apache.iotdb.session.subscription.consumer.tree.SubscriptionTreePullConsumerBuilder;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessage;
import org.apache.iotdb.session.subscription.payload.SubscriptionMessageType;
import org.apache.iotdb.session.subscription.payload.SubscriptionTsFileHandler;
import org.apache.tsfile.read.v4.ITsFileTreeReader;

public class TsFileSubscriptionExample {

  public static void main(String[] args) throws Exception {
    try (ISubscriptionTreePullConsumer consumer =
        new SubscriptionTreePullConsumerBuilder()
            .host("127.0.0.1")
            .port(6667)
            .username("root")
            .password("TimechoDB@2021")
            .consumerId("c1")
            .consumerGroupId("cg1")
            .fileSaveDir("/Users/iotdb/Downloads/subscription-cache")
            .autoCommit(false)
            .build()) {

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

      while (true) {
        final List<SubscriptionMessage> messages = consumer.poll(10_000L);
        for (final SubscriptionMessage message : messages) {
          if (message.getMessageType() != SubscriptionMessageType.TS_FILE.getType()) {
            continue;
          }

          final SubscriptionTsFileHandler handler = message.getTsFile();
          try (ITsFileTreeReader reader = handler.openTreeReader()) {
            System.out.println(reader.getAllDeviceIds());
          }

          final Path archiveDir = Paths.get("/Users/iotdb/Downloads/archive");
          Files.createDirectories(archiveDir);
          final Path target = archiveDir.resolve(handler.getFile().getName());
          handler.copyFile(target);
        }

        if (!messages.isEmpty()) {
          consumer.commitSync(messages);
        }
      }
    }
  }
}

Builder parameter	Default	Description
`host`	`127.0.0.1`	DataNode RPC host.
`port`	`6667`	DataNode RPC port.
`nodeUrls`	`127.0.0.1:6667`	DataNode RPC endpoints. When supplied together with host and port, their union is used.
`username`	`root`	Username.
`password`	`TimechoDB@2021`	Password. Before V2.0.6.x, the default is `root`.
`encryptedPassword`	None	Encrypted password.
`consumerGroupId`	Automatically assigned	Consumer Group ID.
`consumerId`	Automatically assigned	Consumer ID.
`ownerId`	None	Topic Owner ID for advanced configuration.
`ownerEpoch`	None	Topic Owner Epoch for advanced configuration.
`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.
`thriftMaxFrameSize`	Determined by the SDK	Maximum Thrift frame size.
`connectionTimeoutInMs`	`0`	Connection timeout. `0` uses the SDK default behavior.
`maxPollParallelism`	`1`	Maximum parallel poll count.

3.1.2 Pull Consumer Configuration

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

3.1.3 Push Consumer Configuration

Builder parameter	Default	Description
`ackStrategy`	`AckStrategy.AFTER_CONSUME`	Supports `BEFORE_CONSUME` and `AFTER_CONSUME`.
`consumeListener`	Always returns `SUCCESS`	Consumption callback. Business code normally provides it explicitly.
`autoPollIntervalMs`	`100`, minimum `1`	Automatic poll interval in milliseconds.
`autoPollTimeoutMs`	`10000`, minimum `1000`	Timeout for each poll in milliseconds.

3.2 Methods

3.2.1 Topic Management Session

ISubscriptionTreeSession provides the following capabilities:

Method	Description	Return value
`open()`	Opens the Session.	`void`
`createTopic(String topicName)`	Creates a Topic with the default configuration.	`void`
`createTopic(String topicName, Properties config)`	Creates a Topic with the specified configuration.	`void`
`createTopicIfNotExists(...)`	Creates a Topic when it does not exist.	`void`
`alterTopic(String, Properties)`	Modifies Topic configuration.	`void`
`alterTopicOwner(...)`	Modifies the Topic Owner.	`void`
`dropTopic(String topicName)`	Drops a Topic.	`void`
`dropTopicIfExists(String topicName)`	Drops a Topic when it exists.	`void`
`getTopics()`	Gets all Topics.	`Set<Topic>`
`getTopic(String topicName)`	Gets a Topic, or an empty value when it does not exist.	`Optional<Topic>`
`getSubscriptions()`	Gets all subscription relationships.	`Set<Subscription>`
`getSubscriptions(String topicName)`	Gets subscription relationships for a Topic.	`Set<Subscription>`
`dropSubscription(String)`	Drops a subscription relationship.	`void`
`dropSubscriptionIfExists(String)`	Drops a subscription relationship when it exists.	`void`
`close()`	Closes the Session.	`void`

3.2.2 ISubscriptionTreePullConsumer

Method	Description
`open()` / `close()`	Opens or closes the Consumer.
`subscribe(String/String.../Set<String>)`	Subscribes to one or more Topics.
`unsubscribe(String/String.../Set<String>)`	Unsubscribes from one or more Topics.
`poll(Duration/long)`	Pulls messages from the currently subscribed Topics.
`poll(Set<String>, Duration/long)`	Pulls messages from the specified Topics.
`drainBufferedMessages()`	Retrieves messages already buffered by the client.
`commitSync(...)`	Commits one or more messages synchronously.
`commitAsync(...)`	Commits one or more messages asynchronously, optionally with a callback.
`seekToBeginning(String)`	Moves the Topic consumption position to the beginning.
`seekToEnd(String)`	Moves the Topic consumption position to the end.
`positions(String)`	Gets the current consumption position.
`committedPositions(String)`	Gets the committed consumption position.
`seek(String, TopicProgress)`	Moves to the specified consumption position.
`seekAfter(String, TopicProgress)`	Moves to the position after the specified progress.
`getConsumerId()`	Gets the Consumer ID.
`getConsumerGroupId()`	Gets the Consumer Group ID.
`allTopicMessagesHaveBeenConsumed()`	Tests whether all messages in the current Topics have been consumed.

3.2.3 ISubscriptionTreePushConsumer

Method or Builder setting	Description
`open()` / `close()`	Opens or closes the Consumer.
`subscribe(...)` / `unsubscribe(...)`	Subscribes to or unsubscribes from one or more Topics.
`ackStrategy(AckStrategy)`	Configures the acknowledgment strategy.
`consumeListener(ConsumeListener)`	Configures the consumption callback.
`autoPollIntervalMs(long)`	Configures the automatic poll interval.
`autoPollTimeoutMs(long)`	Configures the timeout for each poll.

3.2.4 SubscriptionMessage

SubscriptionMessage is the basic message unit obtained by a Consumer.

Method	Description	Return value
`getMessageType()`	Gets the message type.	`short`
`getResultSets()`	Gets Record ResultSets.	`List<ResultSet>`
`getRecordTabletIterator()`	Gets the Record Tablet iterator.	`Iterator<Tablet>`
`getTsFile()`	Gets the TsFile Handler.	`SubscriptionTsFileHandler`
`getCommitContext()`	Gets the message commit context.	`SubscriptionCommitContext`

3.2.5 SubscriptionTsFileHandler

Method	Description	Return value
`getFile()`	Gets the local temporary file.	`File`
`getPath()`	Gets the local temporary file path.	`Path`
`openTreeReader()`	Opens a tree-model TsFile Reader.	`ITsFileTreeReader`
`openTableReader()`	Opens a table-model TsFile Reader.	`ITsFileReader`
`copyFile(String/Path)`	Copies the subscribed file to a target path.	`Path`
`moveFile(String/Path)`	Moves the subscribed file to a target path.	`Path`
`deleteFile()`	Deletes the local subscription file.	`Path`

4. Connection Timeout and Resource Cleanup

The server uses heartbeats to detect whether a Consumer remains active.
The server can disconnect a Consumer after a long period of inactivity.
When the server disconnects a Consumer, it also triggers unsubscription, clears the subscription relationship, and releases resources.
Use try-with-resources or call close() in a finally block so Sessions and Consumers exit normally.

Data Subscription API

Data Subscription API

1. Core Steps

2. Detailed Steps

2.1 Create a Maven Project

2.2 Examples

2.2.1 Topic Operations

2.2.2 Consume Records in Pull Mode

2.2.3 Commit Consumption Progress Manually

2.2.4 Consume in Push Mode

3. Common API Reference

3.1 Parameters

3.1.1 Common Consumer Configuration

3.1.2 Pull Consumer Configuration

3.1.3 Push Consumer Configuration

3.2 Methods

3.2.1 Topic Management Session

3.2.2 ISubscriptionTreePullConsumer

3.2.3 ISubscriptionTreePushConsumer

3.2.4 SubscriptionMessage

3.2.5 SubscriptionTsFileHandler

4. Connection Timeout and Resource Cleanup