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.

• TunnelWorker requires a warm-up period for initialization, which is specified by the heartbeatIntervalInSec parameter in TunnelWorkerConfig. You can use the setHeartbeatIntervalInSec method in TunnelWorkerConfig to configure this parameter. Default value: 30. Unit: seconds.

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

• The retention period of incremental logs in tunnels is the same as the retention period of Stream logs. Stream logs can be retained for up to seven days. Therefore, incremental logs in tunnels can be retained for up to seven days.

• If you create a tunnel to consume differential or incremental data, take note of the following items:

  • During full data consumption, if the tunnel fails to complete consumption of full data within the retention period of incremental logs (seven days at most), an OTSTunnelExpired error occurs when the tunnel starts to consume incremental logs. As a result, the tunnel cannot consume incremental logs.

    If you estimate that the tunnel cannot complete full data consumption within the specified time window, contact Tablestore technical support.

  • During incremental data consumption, if the tunnel fails to complete consumption of incremental logs within the retention period of incremental logs (seven days at most), the tunnel may consume data from the latest available data. In this case, specific data may not be consumed.

• After a tunnel expires, Tablestore may disable the tunnel. If a tunnel remains in the disabled state for more than 30 days, the tunnel is deleted. You cannot restore a deleted 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

• The following operations are performed in the Resource Access Management (RAM) console:

  • A RAM user is created and the AliyunOTSFullAccess policy is attached to the RAM user to grant the RAM user the permissions to manage Tablestore. For more information, see Create a RAM user and Grant permissions to a RAM user.

    Note

    In actual business environments, we recommend that you grant only the required permissions to RAM users based on the principle of least privilege. This helps prevent security risks caused by excessive user permissions.

  • An AccessKey pair is created for the RAM user. For more information, see Create an AccessKey pair.

    Warning

    If the AccessKey pair of your Alibaba Cloud account is leaked, all resources within the account are exposed to potential risks. We recommend that you use the AccessKey pair of a RAM user to perform operations. This prevents leakage of the AccessKey pair of your Alibaba Cloud account.

• The following operations are performed in the Tablestore console:

  • A data table is created. For more information, see Operations on a data table, Operations on a data table, and Use Tablestore SDKs to create a data table.

  • An endpoint of a Tablestore instance is obtained. For more information, see Obtain an endpoint of a Tablestore instance.

  • Access credentials are configured. For more information, see Configure access credentials.

Use Tunnel Service

In this example, Tablestore SDK for Java is used to get started with Tunnel Service.

1. Initialize a TunnelClient instance.

   Note

   Make sure that the TABLESTORE_ACCESS_KEY_ID and TABLESTORE_ACCESS_KEY_SECRET environment variables are configured. The TABLESTORE_ACCESS_KEY_ID environment variable specifies the AccessKey ID of your Alibaba Cloud account or RAM user. The TABLESTORE_ACCESS_KEY_SECRET environment variable specifies the AccessKey secret of your Alibaba Cloud account or RAM user.

   // 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("TABLESTORE_ACCESS_KEY_ID");
   final String accessKeySecret = System.getenv("TABLESTORE_ACCESS_KEY_SECRET");
   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 in the Tablestore console or by using the createTable method of a SyncClient.

   Important

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

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

   • If you specify the start and end timestamps for incremental data, the valid values must fall within the range of [Current system time - Stream validity period + 5 minutes, Current system time]. Unit: milliseconds.

     • The Stream validity period refers to 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. After you specify the Stream validity period, you cannot modify the period.

     • The end timestamp must be later 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 different type of tunnel, configure the TunnelType parameter in CreateTunnelRequest based on your business 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 data consumption 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 configure custom parameters for a tunnel client based on your business requirements. The following table describes the parameters in Tablestore SDK for Java.

Appendix: 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. Before you perform this step, you must create a table for testing. You can create a table in the Tablestore console or by using the createTable method of a SyncClient. 
        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();
        }
    }
}