Tablestore:Getting started

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.

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 Use the Tablestore console, Use the Tablestore CLI, and Use Tablestore SDKs.

  • The endpoint of the instance in which the data table resides is obtained. For more information, see Initialize a Tablestore client.

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

Get started with Tunnel Service

Use Tablestore SDK for Java to get started with Tunnel Service.

1. Initialize a tunnel client.

   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.

   // Specify the name of the Tablestore instance. 
   // Specify the endpoint of the Tablestore instance. Example: https://instance.cn-hangzhou.ots.aliyuncs.com. 
   // Specify the AccessKey ID and AccessKey secret of your Alibaba Cloud account or RAM user. 
   final String instanceName = "yourInstanceName";
   final String endPoint = "yourEndpoint";
   final String accessKeyId = System.getenv("TABLESTORE_ACCESS_KEY_ID");
   final String accessKeySecret = System.getenv("TABLESTORE_ACCESS_KEY_SECRET");
   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 a Stream or BaseAndStream tunnel, comply with the following rules to specify the timestamps:

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

   • If you specify the start timestamp and the end timestamp for the incremental data, the start timestamp or the end timestamp must fall within the range of [Current system time - Stream validity period + 5 minutes, Current system time] in 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 greater than the start timestamp.

   // You can create three types of tunnels: TunnelType.BaseData, TunnelType.Stream, and TunnelType.BaseAndStream. 
   // The following code provides an example on how to create a BaseAndStream tunnel. To create a tunnel of a different type, set TunnelType in CreateTunnelRequest to the desired type. 
   final String tableName = "testTable";
   final String tunnelName = "testTunnel";
   CreateTunnelRequest request = new CreateTunnelRequest(tableName, tunnelName, TunnelType.BaseAndStream);
   CreateTunnelResponse resp = tunnelClient.createTunnel(request);
   // Use tunnelId to initialize TunnelWorker. 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. The following table describes the configurations of the TunnelClient.

   // 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 use a single server, multiple TunnelWorkers are started. 
   // We recommend that you configure TunnelWorkers by using the same TunnelWorkerConfig. 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 to 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 the tunnel client. 
        // Specify the name of the instance.
        final String instanceName = "yourInstanceName";
        // Specify the endpoint of the instance.
        final String endPoint = "yourEndpoint";
        // Obtain the AccessKey ID and AccessKey secret from the environment variables.
        final String accessKeyId = System.getenv("TABLESTORE_ACCESS_KEY_ID");
        final String accessKeySecret = System.getenv("TABLESTORE_ACCESS_KEY_SECRET");
        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. Specify a custom callback function 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();
        }
    }
}