Managing Consumers

TDengine provides data subscription and consumption interfaces similar to those of message queue products. In many scenarios, by adopting TDengine's time-series big data platform, there is no need to integrate additional message queue products, thus simplifying application design and reducing maintenance costs. This chapter introduces the related APIs and usage methods for data subscription with various language connectors. For basic information on data subscription, please refer to Data Subscription

Creating Topics

Please use taos shell or refer to the Execute SQL section to execute the SQL for creating topics: CREATE TOPIC IF NOT EXISTS topic_meters AS SELECT ts, current, voltage, phase, groupid, location FROM meters

The above SQL will create a subscription named topic_meters. Each record in the messages obtained using this subscription is composed of the columns selected by this query statement SELECT ts, current, voltage, phase, groupid, location FROM meters.

Note In the implementation of TDengine connectors, there are the following limitations for subscription queries.

Query statement limitation: Subscription queries can only use select statements and do not support other types of SQL, such as subscribing to databases, subscribing to supertables (non-select methods), insert, update, or delete, etc.
Raw data query: Subscription queries can only query raw data, not aggregated or calculated results.
Time order limitation: Subscription queries can only query data in chronological order.

Creating Consumers

The concept of TDengine consumers is similar to Kafka, where consumers receive data streams by subscribing to topics. Consumers can be configured with various parameters, such as connection methods, server addresses, automatic Offset submission, etc., to suit different data processing needs. Some language connectors' consumers also support advanced features such as automatic reconnection and data transmission compression to ensure efficient and stable data reception.

Creation Parameters

There are many parameters for creating consumers, which flexibly support various connection types, Offset submission methods, compression, reconnection, deserialization, and other features. The common basic configuration items applicable to all language connectors are shown in the following table:

Parameter Name	Type	Description	Remarks
`td.connect.ip`	string	Server IP address
`td.connect.user`	string	Username
`td.connect.pass`	string	Password
`td.connect.port`	integer	Server port number
`group.id`	string	Consumer group ID, the same consumer group shares consumption progress	Required. Maximum length: 192. Each topic can have up to 100 consumer groups
`client.id`	string	Client ID	Maximum length: 192
`auto.offset.reset`	enum	Initial position of the consumer group subscription	`earliest`: default(version < 3.2.0.0); subscribe from the beginning; `latest`: default(version >= 3.2.0.0); only subscribe from the latest data; `none`: cannot subscribe without a committed offset
`enable.auto.commit`	boolean	Whether to enable automatic consumption point submission, true: automatic submission, client application does not need to commit; false: client application needs to commit manually	Default is true
`auto.commit.interval.ms`	integer	Time interval for automatically submitting consumption records, in milliseconds	Default is 5000
`msg.with.table.name`	boolean	Whether to allow parsing the table name from the message, not applicable to column subscription (column subscription can write tbname as a column in the subquery statement) (from version 3.2.0.0 this parameter is deprecated, always true)	Default is off
`enable.replay`	boolean	Whether to enable data replay function	Default is off
`session.timeout.ms`	integer	Timeout after consumer heartbeat is lost, after which rebalance logic is triggered, and upon success, that consumer will be removed (supported from version 3.3.3.0)	Default is 12000, range [6000, 1800000]
`max.poll.interval.ms`	integer	The longest time interval for consumer poll data fetching, exceeding this time will be considered as the consumer being offline, triggering rebalance logic, and upon success, that consumer will be removed (supported from version 3.3.3.0)	Default is 300000, range [1000, INT32_MAX]

Below are the connection parameters for connectors in various languages:

The parameters for creating a consumer with the Java connector are Properties. For a list of parameters you can set, please refer to Consumer Parameters
For other parameters, refer to the common basic configuration items mentioned above.

The td.connect.websocket.scheme parameter is provided to indicate the protocol type, other parameters are the same as the common basic configuration items.

Supported properties list for creating consumers:

ws.url: WebSocket connection address.
ws.message.channelLen: WebSocket message channel buffer length, default 0.
ws.message.timeout: WebSocket message timeout, default 5m.
ws.message.writeWait: WebSocket message write timeout, default 10s.
ws.message.enableCompression: Whether to enable compression for WebSocket, default false.
ws.autoReconnect: Whether WebSocket should automatically reconnect, default false.
ws.reconnectIntervalMs: WebSocket reconnect interval in milliseconds, default 2000.
ws.reconnectRetryCount: WebSocket reconnect retry count, default 3.

See the table above for other parameters.

The WS_URL parameter is provided to indicate the server address to connect to, other parameters are the same as the common basic configuration items.

Supported properties list for creating consumers:

useSSL: Whether to use SSL connection, default false.
token: Token for connecting to TDengine cloud.
ws.message.enableCompression: Whether to enable WebSocket compression, default false.
ws.autoReconnect: Whether to automatically reconnect, default false.
ws.reconnect.retry.count: Reconnect attempts, default 3.
ws.reconnect.interval.ms: Reconnect interval in milliseconds, default 2000.

See the table above for other parameters.

WebSocket connection: Since it uses dsn, the four configuration items td.connect.ip, td.connect.port, td.connect.user, and td.connect.pass are not needed, the rest are the same as the common configuration items.
Native connection: Same as the common basic configuration items.

WebSocket Connection

Introduces how connectors in various languages use WebSocket connection method to create consumers. Specify the server address to connect, set auto-commit, start consuming from the latest message, specify group.id and client.id, etc. Some language connectors also support deserialization parameters.

Properties config = new Properties();
config.setProperty("td.connect.type", "ws");
config.setProperty("bootstrap.servers", "localhost:6041");
config.setProperty("auto.offset.reset", "latest");
config.setProperty("msg.with.table.name", "true");
config.setProperty("enable.auto.commit", "true");
config.setProperty("auto.commit.interval.ms", "1000");
config.setProperty("group.id", "group1");
config.setProperty("client.id", "clinet1");
config.setProperty("td.connect.user", "root");
config.setProperty("td.connect.pass", "taosdata");
config.setProperty("value.deserializer", "com.taos.example.WsConsumerLoopFull$ResultDeserializer");
config.setProperty("value.deserializer.encoding", "UTF-8");

try {
    TaosConsumer<ResultBean> consumer= new TaosConsumer<>(config);
    System.out.printf("Create consumer successfully, host: %s, groupId: %s, clientId: %s%n",
            config.getProperty("bootstrap.servers"),
            config.getProperty("group.id"),
            config.getProperty("client.id"));
    return consumer;
} catch (Exception ex) {
    // please refer to the JDBC specifications for detailed exceptions info
    System.out.printf("Failed to create websocket consumer, host: %s, groupId: %s, clientId: %s, %sErrMessage: %s%n",
            config.getProperty("bootstrap.servers"),
            config.getProperty("group.id"),
            config.getProperty("client.id"),
            ex instanceof SQLException ? "ErrCode: " + ((SQLException) ex).getErrorCode() + ", " : "",
            ex.getMessage());
    // Print stack trace for context in examples. Use logging in production.
    ex.printStackTrace();
    throw ex;
}

Creating Topics​

Creating Consumers​

Creation Parameters​

WebSocket Connection​

Native Connection​

Subscribe to Consume Data​

WebSocket Connection​

Native Connection​

Specifying the Subscription Offset​

WebSocket Connection​

Native Connection​

Commit Offset​

WebSocket Connection​

Native Connection​

Unsubscribe and Close Consumption​

WebSocket Connection​

Creating Topics

Creating Consumers

Creation Parameters

WebSocket Connection

Native Connection

Subscribe to Consume Data

WebSocket Connection

Native Connection

Specifying the Subscription Offset

WebSocket Connection

Native Connection

Commit Offset

WebSocket Connection

Native Connection

Unsubscribe and Close Consumption

WebSocket Connection