Bugfix: TunnelWorker heartbeat timeout cannot auto retry

okingniko · okingniko · commit 60878aa99f57 · 2019-04-03T10:59:32.000+08:00
diff --git a/pom.xml b/pom.xml
@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.aliyun.openservices</groupId>
     <artifactId>tablestore</artifactId>
-    <version>4.10.1-SNAPSHOT</version>
+    <version>4.10.2-SNAPSHOT</version>
     <packaging>jar</packaging>
     <name>AliCloud TableStore SDK for Java</name>
     <url>http://www.aliyun.com</url>
@@ -183,6 +183,7 @@
                             <exclude>com.lmax:disruptor:jar:</exclude>
                             <exclude>commons-codec:commons-codec:jar:</exclude>
                             <exclude>org.hamcrest:hamcrest-core:jar:</exclude>
+                            <exclude>com.google.code.gson:gson:jar:</exclude>
                         </excludes>
                     </artifactSet>
                     <relocations>
diff --git a/src/main/java/com/alicloud/openservices/tablestore/tunnel/worker/Checkpointer.java b/src/main/java/com/alicloud/openservices/tablestore/tunnel/worker/Checkpointer.java
@@ -57,7 +57,7 @@ public void checkpoint(String token) {
                 }
             }
         } catch (Exception e) {
-            LOG.warn("Checkpoint occurs error, detail: {}", e);
+            LOG.warn("Checkpoint occurs error, detail: {}", e.toString());
         }
     }
 
diff --git a/src/main/java/com/alicloud/openservices/tablestore/tunnel/worker/DefaultChannelProcessor.java b/src/main/java/com/alicloud/openservices/tablestore/tunnel/worker/DefaultChannelProcessor.java
@@ -34,14 +34,14 @@ public void process(ProcessRecordsInput input) {
             try {
                 checkpointer.checkpoint(FINISH_TAG);
             } catch (Exception e) {
-                LOG.error("checkpoint error, detail: {}", e);
+                LOG.error("checkpoint error, detail: {}", e.toString());
             }
         } else if (System.currentTimeMillis() - latestCheckpoint > checkpointIntervalInMillis) {
             LOG.info("begin do checkpoint, token = {}", input.getNextToken());
             try {
                 checkpointer.checkpoint(input.getNextToken());
             } catch (Exception e) {
-                LOG.error("checkpoint error, detail: {}", e);
+                LOG.error("checkpoint error, detail: {}", e.toString());
             }
             latestCheckpoint = System.currentTimeMillis();
         }
diff --git a/src/main/java/com/alicloud/openservices/tablestore/tunnel/worker/TunnelWorker.java b/src/main/java/com/alicloud/openservices/tablestore/tunnel/worker/TunnelWorker.java
@@ -149,6 +149,7 @@ public void run() {
             // 若Worker处于ENDED状态，则需要重新进行资源初始化。
             if (workerStatus.get().equals(TunnelWorkerStatus.WORKER_ENDED)) {
                 workerStatus.set(TunnelWorkerStatus.WORKER_READY);
+                lastHeartbeatTime = new Date();
                 connect();
             }
             try {
diff --git a/src/test/java/com/alicloud/openservices/tablestore/tunnel/functiontest/TestTunnelWorker.java b/src/test/java/com/alicloud/openservices/tablestore/tunnel/functiontest/TestTunnelWorker.java
@@ -1,6 +1,13 @@
 package com.alicloud.openservices.tablestore.tunnel.functiontest;
 
 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.DeleteTunnelRequest;
+import com.alicloud.openservices.tablestore.model.tunnel.DescribeTunnelRequest;
+import com.alicloud.openservices.tablestore.model.tunnel.DescribeTunnelResponse;
+import com.alicloud.openservices.tablestore.model.tunnel.TunnelStage;
+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;
@@ -11,6 +18,7 @@ public class TestTunnelWorker {
     private static final String AccessId = "";
     private static final String AccessKey = "";
     private static final String InstanceName = "";
+    private static final String TableName = "";
 
     static class SimpleProcessor implements IChannelProcessor {
         @Override
@@ -35,23 +43,46 @@ public void shutdown() {
     public static void main(String[] args) {
         TunnelClient client = new TunnelClient(Endpoint, AccessId, AccessKey, InstanceName);
         TunnelWorkerConfig config = new TunnelWorkerConfig(new SimpleProcessor());
-        TunnelWorker worker = new TunnelWorker("480e3a3b-6672-46b1-b970-e611232dcd53", client, config);
+        String tunnelName1 = "test-tunnel-1";
+        CreateTunnelResponse createTunnelResponse =
+            client.createTunnel(new CreateTunnelRequest(TableName, tunnelName1, TunnelType.BaseData));
+        String tunnelId1 = createTunnelResponse.getTunnelId();
+        TunnelWorker worker1 = new TunnelWorker(tunnelId1, client, config);
         try {
             System.out.println("worker running....");
-            worker.connectAndWorking();
-            Thread.sleep(50000);
-            worker.shutdown();
-            System.out.println("worker shutdown....");
+            worker1.connectAndWorking();
+            // 这里会等待一段时间，保证测试的全量可以消费完，具体实现时，可以后台启一个线程进行定期的查询和判断。
+            Thread.sleep(120000);
+            DescribeTunnelResponse resp = client.describeTunnel(new DescribeTunnelRequest(TableName, tunnelName1));
+            if (TunnelStage.ProcessStream.equals(resp.getTunnelInfo().getStage())) {
+                System.out.println("Base data consume finished, shutdown worker.");
+                worker1.shutdown();
+                System.out.println("Delete Tunnel " + tunnelId1);
+                client.deleteTunnel(new DeleteTunnelRequest(TableName, tunnelName1));
+            }
+            System.out.println("worker1 shutdown....");
         } catch (Exception e) {
             e.printStackTrace();
-            worker.shutdown();
+            worker1.shutdown();
         }
 
-        // Reconstruct and connect.
-        TunnelWorker worker2 = new TunnelWorker("480e3a3b-6672-46b1-b970-e611232dcd53", client, config);
+        // 再创建一个新的Tunnel, 进行消费。
+        String tunnelName2 = "test-tunnel-2";
+        CreateTunnelResponse createTunnelResponse2 =
+            client.createTunnel(new CreateTunnelRequest(TableName, tunnelName2, TunnelType.BaseData));
+        String tunnelId2 = createTunnelResponse2.getTunnelId();
+        TunnelWorker worker2 = new TunnelWorker(tunnelId2, client, config);
         try {
             System.out.println("worker2 running...");
             worker2.connectAndWorking();
+            Thread.sleep(60000);
+
+            worker2.shutdown();
+            client.deleteTunnel(new DeleteTunnelRequest(TableName, tunnelName2));
+
+            // 下面两个资源(内部都有线程池)只有在真正需要退出的时候，再shutdown。如果不小心shutdown的话，也要new一个新的出来。
+            client.shutdown();
+            config.shutdown();
         } catch (Exception e) {
             e.printStackTrace();
             worker2.shutdown();
diff --git a/src/test/java/com/alicloud/openservices/tablestore/tunnel/functiontest/TestTunnelWorkerReconnect.java b/src/test/java/com/alicloud/openservices/tablestore/tunnel/functiontest/TestTunnelWorkerReconnect.java
@@ -1,59 +1,18 @@
 package com.alicloud.openservices.tablestore.tunnel.functiontest;
 
-import java.io.FileInputStream;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Properties;
-
-import com.alicloud.openservices.tablestore.SyncClient;
 import com.alicloud.openservices.tablestore.TunnelClient;
-import com.alicloud.openservices.tablestore.model.Column;
-import com.alicloud.openservices.tablestore.model.ColumnValue;
-import com.alicloud.openservices.tablestore.model.CreateTableRequest;
-import com.alicloud.openservices.tablestore.model.DeleteTableRequest;
-import com.alicloud.openservices.tablestore.model.PrimaryKeyBuilder;
-import com.alicloud.openservices.tablestore.model.PrimaryKeyType;
-import com.alicloud.openservices.tablestore.model.PrimaryKeyValue;
-import com.alicloud.openservices.tablestore.model.PutRowRequest;
-import com.alicloud.openservices.tablestore.model.RowPutChange;
-import com.alicloud.openservices.tablestore.model.TableMeta;
-import com.alicloud.openservices.tablestore.model.TableOptions;
-import com.alicloud.openservices.tablestore.model.tunnel.CreateTunnelRequest;
-import com.alicloud.openservices.tablestore.model.tunnel.CreateTunnelResponse;
-import com.alicloud.openservices.tablestore.model.tunnel.DeleteTunnelRequest;
-import com.alicloud.openservices.tablestore.model.tunnel.DeleteTunnelResponse;
-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;
-import org.junit.Assert;
-import org.junit.BeforeClass;
-import org.junit.Test;
 
 public class TestTunnelWorkerReconnect {
-    private static final String CONF_PATH = "tunnel.config";
-    private static String endpoint = "";
-    private static String accessId = "";
-    private static String accessKey = "";
-    private static String instanceName = "";
-
-    @BeforeClass
-    public static  void loadConfig() {
-        System.out.println("here");
-        try {
-            Properties properties = new Properties();
-            properties.load(new FileInputStream(CONF_PATH));
-            endpoint = properties.getProperty("OtsEndpoint");
-            accessId = properties.getProperty("AccessId");
-            accessKey = properties.getProperty("AccessKey");
-            instanceName = properties.getProperty("InstanceName");
-        } catch (Exception e) {
-            e.printStackTrace();
-        }
-    }
+    private static final String Endpoint = "";
+    private static final String AccessId = "";
+    private static final String AccessKey = "";
+    private static final String InstanceName = "";
 
-    class SimpleProcessor implements IChannelProcessor {
+    static class SimpleProcessor implements IChannelProcessor {
         @Override
         public void process(ProcessRecordsInput input) {
             System.out.println("Default record processor, would print records count");
@@ -73,100 +32,16 @@ public void shutdown() {
         }
     }
 
-    static void createTunnel(TunnelClient client, String tableName, String tunnelName) {
-        CreateTunnelRequest request = new CreateTunnelRequest(tableName, tunnelName, TunnelType.Stream);
-        CreateTunnelResponse resp = client.createTunnel(request);
-        System.out.println("RequestId: " + resp.getRequestId());
-        System.out.println("TunnelId: " + resp.getTunnelId());
-    }
-
-    static void createTable(SyncClient client, String tableName) {
-        TableMeta meta = new TableMeta(tableName);
-        meta.addPrimaryKeyColumn("int", PrimaryKeyType.INTEGER);
-        meta.addPrimaryKeyColumn("str", PrimaryKeyType.STRING);
-        CreateTableRequest request = new CreateTableRequest(meta, new TableOptions(-1, 1));
-        client.createTable(request);
-
-    }
-
-    static void deleteTable(SyncClient client, String tableName) {
-        DeleteTableRequest request = new DeleteTableRequest(tableName);
-        client.deleteTable(request);
-    }
-
-    static List<RowPutChange> putRows(SyncClient client, String tableName, int rowCount) {
-        List<RowPutChange> changes = new ArrayList<RowPutChange>();
-        for (int i = 0; i < rowCount; i++) {
-            PrimaryKeyBuilder primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();
-            primaryKeyBuilder.addPrimaryKeyColumn("int", PrimaryKeyValue.fromLong(i));
-            primaryKeyBuilder.addPrimaryKeyColumn("str", PrimaryKeyValue.fromString("string" + i));
-            RowPutChange rowPutChange = new RowPutChange(tableName, primaryKeyBuilder.build());
-
-            for (int j = 0; j < 10; j++) {
-                rowPutChange.addColumn(new Column("test" + j, ColumnValue.fromLong(i)));
-            }
-            client.putRow(new PutRowRequest(rowPutChange));
-            changes.add(rowPutChange);
-        }
-        System.out.println(String.format("Put %d rows succeed.", rowCount));
-        return changes;
-    }
-
-    static void deleteTunnel(TunnelClient client, String tableName, String tunnelName) {
-        DeleteTunnelRequest request = new DeleteTunnelRequest(tableName, tunnelName);
-        DeleteTunnelResponse resp = client.deleteTunnel(request);
-        System.out.println("RequestId: " + resp.getRequestId());
-    }
-
-    @Test
-    public void testTunnelWorkerReconnect_WithTunnelInvalid() {
-        TunnelClient tunnelClient = new TunnelClient(endpoint, accessId, accessKey, instanceName);
-        SyncClient syncClient = new SyncClient(endpoint, accessId, accessKey, instanceName);
-
-        String tableName = "test_zr" + System.currentTimeMillis();
-        String tunnelName = "test_zr" + System.currentTimeMillis();
-        //  1. create table
-        System.out.println("Begin Create Table: " + tableName);
-        createTable(syncClient, tableName);
-        System.out.println("++++++++++++++++++++++++++++++++++++");
-
-        //  2. create tunnel
-        System.out.println("Begin Create Tunnel: " + tunnelName);
-        CreateTunnelResponse resp = tunnelClient.createTunnel(
-            new CreateTunnelRequest(tableName, tunnelName, TunnelType.Stream));
-        String tunnelId = resp.getTunnelId();
-        System.out.println("Create Tunnel, Id: " + tunnelId);
-        System.out.println("++++++++++++++++++++++++++++++++++++");
-
-        //  3. put data
-        System.out.println("Begin Put Data in backend.");
-        putRows(syncClient, tableName, 5000);
-        System.out.println("++++++++++++++++++++++++++++++++++++");
-
-        // 4. new tunnel worker and consume.
+    public static void main(String[] args) {
+        TunnelClient client = new TunnelClient(Endpoint, AccessId, AccessKey, InstanceName);
         TunnelWorkerConfig config = new TunnelWorkerConfig(new SimpleProcessor());
-        TunnelWorker worker = new TunnelWorker(tunnelId, tunnelClient, config);
+        TunnelWorker worker1 = new TunnelWorker("cfea9e1d-19f3-4835-8551-698a5c5c0773", client, config);
         try {
-            worker.connectAndWorking();
-            Thread.sleep(50000);
-            System.out.println("Begin Delete Tunnel: " + tunnelName);
-            deleteTunnel(tunnelClient, tableName, tunnelName);
-            System.out.println("++++++++++++++++++++++++++++++++++++");
-            Thread.sleep(50000);
-            // cannot achieve here.
-            Assert.fail();
+            System.out.println("worker running....");
+            worker1.connectAndWorking();
         } catch (Exception e) {
             e.printStackTrace();
-            worker.shutdown();
-        } finally {
-            //  delete table
-            System.out.println("Begin Delete Table: " + tableName);
-            deleteTable(syncClient, tableName);
-            System.out.println("++++++++++++++++++++++++++++++++++++");
-            worker.shutdown();
-            syncClient.shutdown();
-            tunnelClient.shutdown();
+            worker1.shutdown();
         }
     }
-
 }

Original file line number	Diff line number	Diff line change
`@@ -57,7 +57,7 @@ public void checkpoint(String token) {`
`57`	`57`	`}`
`58`	`58`	`}`
`59`	`59`	`} catch (Exception e) {`
`60`		`- LOG.warn("Checkpoint occurs error, detail: {}", e);`
	`60`	`+ LOG.warn("Checkpoint occurs error, detail: {}", e.toString());`
`61`	`61`	`}`
`62`	`62`	`}`
`63`	`63`
Original file line number	Diff line number	Diff line change
`@@ -34,14 +34,14 @@ public void process(ProcessRecordsInput input) {`
`34`	`34`	`try {`
`35`	`35`	`checkpointer.checkpoint(FINISH_TAG);`
`36`	`36`	`} catch (Exception e) {`
`37`		`- LOG.error("checkpoint error, detail: {}", e);`
	`37`	`+ LOG.error("checkpoint error, detail: {}", e.toString());`
`38`	`38`	`}`
`39`	`39`	`} else if (System.currentTimeMillis() - latestCheckpoint > checkpointIntervalInMillis) {`
`40`	`40`	`LOG.info("begin do checkpoint, token = {}", input.getNextToken());`
`41`	`41`	`try {`
`42`	`42`	`checkpointer.checkpoint(input.getNextToken());`
`43`	`43`	`} catch (Exception e) {`
`44`		`- LOG.error("checkpoint error, detail: {}", e);`
	`44`	`+ LOG.error("checkpoint error, detail: {}", e.toString());`
`45`	`45`	`}`
`46`	`46`	`latestCheckpoint = System.currentTimeMillis();`
`47`	`47`	`}`
Original file line number	Diff line number	Diff line change
`@@ -149,6 +149,7 @@ public void run() {`
`149`	`149`	`// 若Worker处于ENDED状态，则需要重新进行资源初始化。`
`150`	`150`	`if (workerStatus.get().equals(TunnelWorkerStatus.WORKER_ENDED)) {`
`151`	`151`	`workerStatus.set(TunnelWorkerStatus.WORKER_READY);`
	`152`	`+ lastHeartbeatTime = new Date();`
`152`	`153`	`connect();`
`153`	`154`	`}`
`154`	`155`	`try {`