Tablestore:Use Tunnel Service by using Tablestore SDKs

Usage notes

• By default, the system starts a thread pool to read and process data based on TunnelWorkerConfig. If you want to start multiple TunnelWorkers on a single server, we recommend that you use the same TunnelWorkerConfig to configure all TunnelWorkers.

• The incremental logs for a tunnel are retained for a maximum of seven days. The specific storage duration of incremental logs is consistent with that of the stream logs for a data table. If you create a tunnel to consume differential or incremental data, take note of the following items:

  • During full data consumption, if the tunnel does not complete the consumption of full data within seven days, an OTSTunnelExpired error occurs when the tunnel starts to consume incremental data. As a result, the tunnel cannot consume incremental data. If you estimate that full data cannot be consumed within seven days, contact Tablestore technical support.

  • During incremental data consumption, if the tunnel has not consumed incremental data for more than seven days, the consumption starts from the latest data that can be consumed. In this case, some data may not be consumed.

    Note

    If incremental data has not been consumed for more than seven days, the incremental data expires. The specific storage duration of incremental logs is consistent with that of the stream logs for a data table. If the incremental data is expired for the specified period, which is seven days by default, Tablestore disables the tunnel. In this case, the tunnel cannot be used to consume data.

• TunnelWorker requires time to warm up for initialization, which is determined by the heartbeatIntervalInSec parameter in TunnelWorkerConfig. You can use the setHeartbeatIntervalInSec method in TunnelWorkerConfig to configure this parameter. Default value: 30. Minimum value: 5. Unit: seconds.

• When the mode switches from the full channel to the incremental channel, the full channel is closed and the incremental channel is started. This process requires another period of time for initialization, which is also determined by the heartbeatIntervalInSec parameter.

• When the TunnelWorker client is shut down due to an unexpected exit or manual termination, TunnelWorker uses one of the following methods to automatically recycle resources: release the thread pool, automatically call the shutdown method that you have registered for the Channel class, or shut down the tunnel.

API operations

Use Tablestore SDKs

You can use Tablestore SDKs for the following programming languages to implement Tunnel Service:

• Java SDK

• Go SDK

Prerequisites

• An AccessKey pair is created and configured in environment variables. The endpoint that you want to use is obtained. For more information, see Initialize an OTSClient instance.

  Tablestore uses the OTS_AK_ENV and OTS_SK_ENV environment variables to store an AccessKey pair. The OTS_AK_ENV environment variable stores the AccessKey ID of an Alibaba Cloud account or a Resource Access Management (RAM) user. The OTS_SK_ENV environment variable stores the AccessKey secret of an Alibaba Cloud account or a RAM user. Configure the environment variables based on the AccessKey pair that you want to use.

• A data table is created. For more information, see the "Create a data table" section of the Use the Wide Column model in the Tablestore console topic or the Operations on tables topic.

Use Tunnel Service

Tablestore SDK for Java allows you to use Tunnel Service.

1. Initialize a Tunnel client.

   // Set the endPoint parameter to the endpoint of the Tablestore instance. Example: https://instance.cn-hangzhou.ots.aliyuncs.com. 
   // Set the accessKeyId parameter to the AccessKey ID and the accessKeySecret parameter to the AccessKey secret that you want to use to access Tablestore. 
   // Set the instanceName parameter to the name of the instance. 
   final String endPoint = "";
   final String accessKeyId = System.getenv("OTS_AK_ENV");
   final String accessKeySecret = System.getenv("OTS_SK_ENV");
   final String instanceName = "";
   TunnelClient tunnelClient = new TunnelClient(endPoint, accessKeyId, accessKeySecret, instanceName);

2. Create a tunnel.

   Before you create a tunnel, create a data table for testing or prepare an existing table. You can create a table by using the createTable method of a SyncClient or in the Tablestore console.

   Important

   When you create a tunnel of the incremental or differential type, you must comply with the following rules to specify the timestamps:

   • If you do not specify the start timestamp for incremental data, the time when the tunnel was created is used as the start timestamp.

   • If you specify the start timestamp and the end timestamp for incremental data, the valid values of the start timestamp and the end timestamp are in the range of [Current system time - Stream validity period + 5 minutes, Current system time. Unit: milliseconds.

     • The Stream validity period is the validity period of incremental logs in milliseconds. The maximum Stream validity period is seven days. You can specify the Stream validity period when you enable Stream for the data table. You cannot modify the Stream validity period after you specify it.

     • The end timestamp must be greater than the start timestamp.

   // The following types of tunnels are supported: TunnelType.BaseData, TunnelType.Stream, and TunnelType.BaseAndStream. 
   // The following sample code provides an example on how to create a BaseAndStream tunnel. To create a tunnel of a different type, configure the TunnelType parameter in CreateTunnelRequest based on your requirements. 
   final String tableName = "testTable";
   final String tunnelName = "testTunnel";
   CreateTunnelRequest request = new CreateTunnelRequest(tableName, tunnelName, TunnelType.BaseAndStream);
   CreateTunnelResponse resp = tunnelClient.createTunnel(request);
   // Use the tunnelId parameter to initialize a TunnelWorker. You can call the ListTunnel or DescribeTunnel operation to obtain the tunnel ID. 
   String tunnelId = resp.getTunnelId(); 
   System.out.println("Create Tunnel, Id: " + tunnelId);

3. Specify a custom callback to start automatic data consumption.

   // Specify the callback for data consumption to call the IChannelProcessor operation, which specifies the process and shutdown methods. 
   private static class SimpleProcessor implements IChannelProcessor {
       @Override
       public void process(ProcessRecordsInput input) {
           // The ProcessRecordsInput parameter contains the data that you obtain. 
           System.out.println("Default record processor, would print records count");
           System.out.println(
               // The NextToken parameter is used by the Tunnel client to paginate data. 
               String.format("Process %d records, NextToken: %s", input.getRecords().size(), input.getNextToken()));
           try {
               // Simulate data consumption and processing. 
               Thread.sleep(1000);
           } catch (InterruptedException e) {
               e.printStackTrace();
           }
       }
       @Override
       public void shutdown() {
           System.out.println("Mock shutdown");
       }
   }
   
   // By default, the system starts a thread pool to read and process data based on TunnelWorkerConfig. 
   // If you want to start multiple TunnelWorkers on a single server, we recommend that you use the same TunnelWorkerConfig to configure all TunnelWorkers. TunnelWorkerConfig provides more advanced parameters. 
   TunnelWorkerConfig config = new TunnelWorkerConfig(new SimpleProcessor());
   // Configure TunnelWorkers and start automatic data processing. 
   TunnelWorker worker = new TunnelWorker(tunnelId, tunnelClient, config);
   try {
       worker.connectAndWorking();
   } catch (Exception e) {
       e.printStackTrace();
       config.shutdown();
       worker.shutdown();
       tunnelClient.shutdown();
   }

Configure TunnelWorkerConfig

TunnelWorkerConfig allows you to customize parameters for a TunnelClient instance based on your requirements. The following table describes the parameters.

Appendix: Complete sample code

import com.alicloud.openservices.tablestore.TunnelClient;
import com.alicloud.openservices.tablestore.model.tunnel.CreateTunnelRequest;
import com.alicloud.openservices.tablestore.model.tunnel.CreateTunnelResponse;
import com.alicloud.openservices.tablestore.model.tunnel.TunnelType;
import com.alicloud.openservices.tablestore.tunnel.worker.IChannelProcessor;
import com.alicloud.openservices.tablestore.tunnel.worker.ProcessRecordsInput;
import com.alicloud.openservices.tablestore.tunnel.worker.TunnelWorker;
import com.alicloud.openservices.tablestore.tunnel.worker.TunnelWorkerConfig;
public class TunnelQuickStart {
    private static class SimpleProcessor implements IChannelProcessor {
        @Override
        public void process(ProcessRecordsInput input) {
            System.out.println("Default record processor, would print records count");
            System.out.println(
                // The NextToken parameter is used by the Tunnel client to paginate data. 
                String.format("Process %d records, NextToken: %s", input.getRecords().size(), input.getNextToken()));
            try {
                // Simulate data consumption and processing. 
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        @Override
        public void shutdown() {
            System.out.println("Mock shutdown");
        }
    }
    public static void main(String[] args) throws Exception {
        //1. Initialize a Tunnel client. 
        final String endPoint = "";
        final String accessKeyId = System.getenv("OTS_AK_ENV");
        final String accessKeySecret = System.getenv("OTS_SK_ENV");
        final String instanceName = "";
        TunnelClient tunnelClient = new TunnelClient(endPoint, accessKeyId, accessKeySecret, instanceName);
        //2. Create a tunnel. You must create a table for testing before you perform this step. You can create a table by using the createTable method of a SyncClient or in the Tablestore console. 
        final String tableName = "testTable";
        final String tunnelName = "testTunnel";
        CreateTunnelRequest request = new CreateTunnelRequest(tableName, tunnelName, TunnelType.BaseAndStream);
        CreateTunnelResponse resp = tunnelClient.createTunnel(request);
        // Use the tunnelId parameter to initialize a TunnelWorker. You can call the ListTunnel or DescribeTunnel operation to obtain the tunnel ID. 
        String tunnelId = resp.getTunnelId();
        System.out.println("Create Tunnel, Id: " + tunnelId);
        //3. Define a custom callback to start automatic data consumption. 
        // TunnelWorkerConfig provides more advanced parameters. 
        TunnelWorkerConfig config = new TunnelWorkerConfig(new SimpleProcessor());
        TunnelWorker worker = new TunnelWorker(tunnelId, tunnelClient, config);
        try {
            worker.connectAndWorking();
        } catch (Exception e) {
            e.printStackTrace();
            config.shutdown();
            worker.shutdown();
            tunnelClient.shutdown();
        }
    }
}