Create a tunnel - Tablestore SDK for Java - Tablestore - Alibaba Cloud Documentation Center

Prerequisites

Feature description

Call createTunnel to create a tunnel for a data table. You can create multiple tunnels for the same data table. The tunnel type determines the data to consume: BaseData consumes only full data, Stream consumes only incremental data, and BaseAndStream consumes full data and then incremental data.

Note

If Stream is disabled for a data table, creating a Stream or BaseAndStream tunnel automatically enables Stream and sets the retention period of incremental logs to seven days.

CreateTunnelResponse createTunnel(CreateTunnelRequest request)
        throws TableStoreException, ClientException

The following sample code creates a BaseData tunnel named example_tunnel for the example_table data table.

String tableName = "example_table";
String tunnelName = "example_tunnel";

CreateTunnelRequest request =
        new CreateTunnelRequest(
                tableName, tunnelName, TunnelType.BaseData);
CreateTunnelResponse response = tunnelClient.createTunnel(request);

System.out.println("TunnelId: " + response.getTunnelId());

Parameters

CreateTunnelRequest contains the following parameters:

Name	Type	Description
tableName (required)	`String`	The name of the data table.
tunnelName (required)	`String`	The tunnel name.
tunnelType (required)	`TunnelType`	The tunnel type. `BaseData` consumes only full data. `Stream` consumes only incremental data. `BaseAndStream` consumes full data and then incremental data.
streamTunnelConfig (optional)	`StreamTunnelConfig`	The incremental data range configuration for `Stream` and `BaseAndStream` tunnels. If you configure this parameter, specify a valid time range.
streamRecordOptions (optional)	`StreamRecordOptions`	The incremental record configuration for `Stream` and `BaseAndStream` tunnels. This parameter requires Tablestore SDK for Java 5.17.11 or later.

Incremental data range

streamTunnelConfig is of the StreamTunnelConfig type and contains the following parameters:

Name	Type	Description
flag (optional)	`StartOffsetFlag`	The start position used when `startOffset` is not specified. `LATEST` starts from the tunnel creation time. `EARLIEST` starts from the earliest incremental log that is available. Default value: `LATEST`.
startOffset (optional)	`long`	The start timestamp of incremental data. Unit: milliseconds. Valid range: [Current system time - Stream retention period + 5 minutes, current system time). If this parameter is specified, `flag` does not take effect.
endOffset (optional)	`long`	The end timestamp of incremental data. Unit: milliseconds. If both timestamps are specified, this parameter must be greater than `startOffset`. If this parameter is not specified, incremental data continues to be consumed.

Note

The Stream retention period is the retention period of incremental logs. The maximum value is seven days. You can set the period when you enable Stream for a data table. The period cannot be modified after it is set.

Incremental record content

streamRecordOptions is of the StreamRecordOptions type and contains the following parameters:

Name	Type	Description
getVersionGeneratorValue (optional)	`boolean`	Specifies whether incremental records contain the value generated by the version generator. Default value: `false`.
getSysColumns (optional)	`boolean`	Specifies whether incremental records contain system columns. Default value: `false`.
getNewRowInfo (optional)	`boolean`	Specifies whether incremental records contain the latest row information. Default value: `false`.
oldColumnsToGet (optional)	`StreamColumn`	The attribute columns to return from the original row.
newColumnsToGet (optional)	`StreamColumn`	The attribute columns to return from the latest row.

Incremental record columns

streamRecordOptions.oldColumnsToGet and streamRecordOptions.newColumnsToGet are of the StreamColumn type and contain the following parameters:

Name	Type	Description
columnType (required)	`StreamColumnType`	The attribute column selection mode. `SPECIFIED_COLUMN` specifies named attribute columns. `INPUT_COLUMNS` specifies the attribute columns written or updated by the operation. `ALL_COLUMNS` specifies all attribute columns.
columnNames (optional)	`List<String>`	The names of the attribute columns to return when `columnType` is set to `SPECIFIED_COLUMN`.

Response

CreateTunnelResponse contains the following response field:

Field	Type	Description
tunnelId	`String`	The ID of the created tunnel. Call `getTunnelId()` to obtain the ID. The ID is required to consume tunnel data.

Scenarios

Specify an incremental data range

The following sample code creates a Stream tunnel and specifies the incremental data range of the last hour.

long endTime = System.currentTimeMillis() - 1_000L;
long startTime = endTime - 60 * 60 * 1000L;
StreamTunnelConfig streamConfig =
        new StreamTunnelConfig(startTime, endTime);

CreateTunnelRequest request =
        new CreateTunnelRequest(
                "example_table",
                "example_stream_tunnel",
                TunnelType.Stream);
request.setStreamTunnelConfig(streamConfig);

CreateTunnelResponse response = tunnelClient.createTunnel(request);
System.out.println("TunnelId: " + response.getTunnelId());

Configure incremental record content

The following sample code creates a Stream tunnel and configures incremental records to return the version generator value, system columns, latest row information, the value attribute column from the original row, and all attribute columns from the latest row.

StreamColumn oldColumns =
        new StreamColumn(StreamColumnType.SPECIFIED_COLUMN);
oldColumns.addColumnName("value");

StreamRecordOptions recordOptions = new StreamRecordOptions();
recordOptions.setGetVersionGeneratorValue(true);
recordOptions.setGetSysColumns(true);
recordOptions.setGetNewRowInfo(true);
recordOptions.setOldColumnsToGet(oldColumns);
recordOptions.setNewColumnsToGet(
        new StreamColumn(StreamColumnType.ALL_COLUMNS));

CreateTunnelRequest request =
        new CreateTunnelRequest(
                "example_table", "example_stream_tunnel", TunnelType.Stream);
request.setStreamRecordOptions(recordOptions);

CreateTunnelResponse response = tunnelClient.createTunnel(request);
System.out.println("TunnelId: " + response.getTunnelId());