Consume data from a tunnel - Tablestore SDK for Java - Tablestore

Usage notes

The retention period of incremental logs is the same as the Stream log expiration period of the table and can be up to seven days. For a BaseAndStream tunnel, if full data consumption does not finish within this period, the OTSTunnelExpired error is returned when incremental data consumption begins. The tunnel cannot continue to consume incremental data.
If incremental consumption falls behind the retention period, the tunnel may resume from the latest available data. As a result, some data may not be consumed.
An expired tunnel may be disabled. If the tunnel remains disabled for more than 30 days, it is deleted and cannot be restored.

Prerequisites

Install the Tablestore SDK for Java and initialize TunnelClient.

Feature description

TunnelWorker connects to a tunnel by tunnel ID, uses heartbeats to obtain the channels assigned to the current client, continuously pulls data, and passes each batch of records to IChannelProcessor. If multiple TunnelWorker instances consume the same tunnel, the server distributes channels among the clients.

To consume data from a tunnel:

Implement IChannelProcessor. Use the process method to process each batch and the shutdown method to release resources used by the callback.
Create a TunnelWorkerConfig object to configure the callback and consumption behavior.
Create a TunnelWorker with the tunnel ID, TunnelClient, and TunnelWorkerConfig.

Call connectAndWorking to start consumption.

void process(ProcessRecordsInput input);
void shutdown();

The following example prints each record pulled from the tunnel and then starts consumption.

private static class SimpleProcessor implements IChannelProcessor {
    @Override
    public void process(ProcessRecordsInput input) {
        for (StreamRecord record : input.getRecords()) {
            System.out.println(record);
        }
    }

    @Override
    public void shutdown() {
        // Release resources used by the callback.
    }
}

String tunnelId = "example_tunnel_id";
TunnelWorkerConfig config =
        new TunnelWorkerConfig(new SimpleProcessor());
TunnelWorker worker =
        new TunnelWorker(tunnelId, tunnelClient, config);
worker.connectAndWorking();

Important

connectAndWorking returns after it starts background consumption tasks. Keep the application process running. To stop consumption, call worker.shutdown(), config.shutdown(), and tunnelClient.shutdown() in that order. worker.shutdown() closes tunnel connections and invokes the callback's shutdown method. config.shutdown() shuts down the read, processing, and helper thread pools. TunnelWorker registers a JVM shutdown hook that attempts to stop the worker, but the application must still explicitly release these resources.

Parameters

Worker

The TunnelWorker constructor contains the following parameters.

Name	Type	Description
tunnelId (required)	String	The tunnel ID. Obtain it by creating, listing, or querying tunnels.
client (required)	TunnelClientInterface	The initialized `TunnelClient`.
workerConfig (required)	TunnelWorkerConfig	The callback and consumption behavior configuration.

Consumption configuration

workerConfig is of the TunnelWorkerConfig type and contains the following parameters.

Name	Type	Description
channelProcessor (required)	IChannelProcessor	The data processing callback. This parameter is required when you use the three-parameter `TunnelWorker` constructor.
heartbeatTimeoutInSec (optional)	long	The heartbeat timeout in seconds. The default value is `300` and must be greater than `heartbeatIntervalInSec`. After a heartbeat times out, the server considers the client unavailable and the client reconnects to the tunnel.
heartbeatIntervalInSec (optional)	long	The heartbeat interval in seconds. The default value is `30`, and the minimum value is `5`. Heartbeats obtain active channels, update channel states, and initialize data processing tasks. The interval also affects `TunnelWorker` warm-up time.
checkpointIntervalInMillis (optional)	long	The interval at which consumption checkpoints are recorded on the server, in milliseconds. The default value is `5000`. Tunnel Service delivers each record at least once and preserves record order. A restarted task resumes from the latest checkpoint, so it may process some data more than once. A shorter interval reduces duplicate processing, but recording checkpoints too frequently can reduce throughput.
clientTag (optional)	String	A custom client tag that is used to generate a client ID and distinguish `TunnelWorker` instances. The default value is the Java `os.name` system property.
readRecordsExecutor (optional)	ThreadPoolExecutor	The thread pool that pulls data. The default pool has `32` core threads, up to `1000` threads, a queue capacity of `16`, and a `60`-second keep-alive time.
processRecordsExecutor (optional)	ThreadPoolExecutor	The thread pool that processes data. Its default configuration is the same as `readRecordsExecutor`. For a custom pool, configure the number of threads based on the number of channels in the tunnel.
maxChannelParallel (optional)	int	The maximum number of channels from which data is concurrently pulled and processed. Use this parameter to limit memory usage. The default value is `-1`, which specifies no limit. Tablestore SDK for Java 5.10.0 and later support this parameter.
channelHelperExecutor (optional)	ThreadPoolExecutor	The helper thread pool that initializes channels, schedules pipelines, and handles runtime errors. If this parameter is not set, a cached thread pool is used.
maxRetryIntervalInMillis (optional)	int	The maximum base interval for exponential backoff during incremental data pulls, in milliseconds. The default value is `2000`, and the minimum value is `200`. If a batch contains no more than 500 records and is no larger than 900 KB, the client gradually increases the backoff interval. The actual interval is randomly selected from 75% to 125% of the current base interval. Tablestore SDK for Java 5.4.0 and later support this parameter.
readMaxTimesPerRound (optional)	int	The maximum number of `ReadRecords` calls in one pipeline round. The default value is `1`.
readMaxBytesPerRound (optional)	int	The maximum amount of data pulled in one pipeline round, in bytes. The default value is `4194304`, or 4 MiB. The round stops when this value or `readMaxTimesPerRound` is reached.
enableClosingChannelDetect (optional)	boolean	Specifies whether to detect channels in the `CLOSING` state in real time. A `CLOSING` channel is being migrated from one client to another. Tablestore SDK for Java 5.13.13 and later support this parameter. The default value is `true` in version 5.17.0 and later. If detection is disabled, channel migration can be blocked and consumption can be interrupted when many channels exist but client resources are insufficient.

If you start multiple TunnelWorker instances on the same machine, you can reuse one TunnelWorkerConfig object to share the read and processing thread pools. After all workers stop, call config.shutdown() only once.

Callback data

The process method receives a ProcessRecordsInput object that contains the following fields.

Field	Type	Description
records	`List<StreamRecord>`	The records pulled in the current batch. Call `getRecords()` to obtain them.
nextToken	String	The token for the next batch. Call `getNextToken()` to obtain it. `TunnelWorker` automatically uses this value to continue pulling data and record checkpoints.
traceId	String	The trace ID of the current pull request. Call `getTraceId()` to obtain it.
channelId	String	The ID of the channel to which the current batch belongs. Call `getChannelId()` to obtain it. Call `getPartitionId()` to obtain the partition ID from the channel ID.

Scenario examples

Tune consumption parameters

If consumption throughput or memory usage does not meet your requirements, adjust the heartbeat and checkpoint intervals, channel concurrency, the number and size of pulls per round, and the backoff interval for incremental pulls.

TunnelWorkerConfig config =
        new TunnelWorkerConfig(new SimpleProcessor());
config.setHeartbeatIntervalInSec(10);
config.setHeartbeatTimeoutInSec(60);
config.setCheckpointIntervalInMillis(10_000);
config.setMaxChannelParallel(16);
config.setReadMaxTimesPerRound(4);
config.setReadMaxBytesPerRound(8 * 1024 * 1024);
config.setMaxRetryIntervalInMillis(3_000);