Tablestore:Consume incremental data

Prerequisites

• Install the Tablestore SDK for Java and initialize the client.

• Stream enabled on the table by setting StreamSpecification at table creation. For details, see Create a data table.

Description

Stream organizes incremental changes from a table into shards. To consume a stream, call four APIs in sequence: list streams, describe a stream, get a shard iterator, and fetch records.

1. Call listStream(ListStreamRequest) to list the streamId values of all Stream-enabled tables in the instance.

2. Call describeStream(DescribeStreamRequest) to retrieve stream metadata (creation time, expiration time, current status) and the list of Shard objects.

3. Call getShardIterator(GetShardIteratorRequest) to get the read iterator (shardIterator) for a specific Shard. The iterator marks where to start fetching incremental records.

4. Call getStreamRecord(GetStreamRecordRequest) with the shardIterator to fetch a batch of incremental records (a list of StreamRecord objects). Use the returned nextShardIterator to fetch subsequent records.

public ListStreamResponse listStream(ListStreamRequest request) throws TableStoreException, ClientException
public DescribeStreamResponse describeStream(DescribeStreamRequest request) throws TableStoreException, ClientException
public GetShardIteratorResponse getShardIterator(GetShardIteratorRequest request) throws TableStoreException, ClientException
public GetStreamRecordResponse getStreamRecord(GetStreamRecordRequest request) throws TableStoreException, ClientException

The following example consumes the stream_test_demo stream end-to-end and prints the type and primary key of each record.

String demoTable = "stream_test_demo";

// 1. List all tables in the instance with Stream enabled and find the streamId of the target table.
ListStreamRequest listRequest = new ListStreamRequest(demoTable);
ListStreamResponse listResponse = client.listStream(listRequest);

String targetStreamId = null;
for (Stream stream : listResponse.getStreams()) {
    if (demoTable.equals(stream.getTableName())) {
        targetStreamId = stream.getStreamId();
        break;
    }
}
System.out.println("Stream ID: " + targetStreamId);

// 2. Query all shards of the Stream.
DescribeStreamRequest describeRequest = new DescribeStreamRequest(targetStreamId);
DescribeStreamResponse describeResponse = client.describeStream(describeRequest);
List<StreamShard> shards = describeResponse.getShards();
System.out.println("Shard count: " + shards.size());

if (!shards.isEmpty()) {
    String shardId = shards.get(0).getShardId();

    // 3. Get the initial read iterator of the shard.
    GetShardIteratorRequest iterRequest =
            new GetShardIteratorRequest(targetStreamId, shardId);
    GetShardIteratorResponse iterResponse = client.getShardIterator(iterRequest);
    String shardIterator = iterResponse.getShardIterator();

    // 4. Use the iterator to pull incremental records from the shard.
    GetStreamRecordRequest recordRequest = new GetStreamRecordRequest(shardIterator);
    recordRequest.setLimit(100);
    GetStreamRecordResponse recordResponse = client.getStreamRecord(recordRequest);

    List<StreamRecord> records = recordResponse.getRecords();
    System.out.println("Records fetched: " + records.size());
    for (StreamRecord record : records) {
        System.out.println("RecordType: " + record.getRecordType()
                + ", PK: " + record.getPrimaryKey());
    }

    // nextShardIterator is used to continue pulling subsequent incremental records.
    System.out.println("Next iterator: "
            + (recordResponse.getNextShardIterator() != null ? "yes" : "no"));
}

Parameters

Response

Examples

String currentStreamId = "<your-stream-id>";
String startShardId = null;
int totalShards = 0;

while (true) {
    DescribeStreamRequest request = new DescribeStreamRequest(currentStreamId);
    if (startShardId != null) {
        request.setInclusiveStartShardId(startShardId);
    }
    request.setShardLimit(50);

    DescribeStreamResponse response = client.describeStream(request);
    totalShards += response.getShards().size();

    // A null nextShardId indicates that all shards have been traversed.
    if (response.getNextShardId() == null) {
        break;
    }
    startShardId = response.getNextShardId();
}
System.out.println("Total shards: " + totalShards);

String currentStreamId = "<your-stream-id>";
String shardId = "<your-shard-id>";

GetShardIteratorRequest iterRequest =
        new GetShardIteratorRequest(currentStreamId, shardId);
String shardIterator = client.getShardIterator(iterRequest).getShardIterator();

int totalRecords = 0;
while (shardIterator != null) {
    GetStreamRecordRequest recordRequest = new GetStreamRecordRequest(shardIterator);
    recordRequest.setLimit(100);
    GetStreamRecordResponse response = client.getStreamRecord(recordRequest);

    totalRecords += response.getRecords().size();
    shardIterator = response.getNextShardIterator();
}
System.out.println("Polling total records: " + totalRecords);