update version to 5.13.13

Author: whybert

README.md

@@ -9,19 +9,19 @@
  - 阿里云表格存储是阿里云自主研发的NoSQL数据存储服务，提供海量结构化数据的存储和实时访问。
 
 ## 版本
- - 当前版本：5.13.12
+ - 当前版本：5.13.13
 
 ## 运行环境
  - JDK 6及其以上
 
 ## 安装
 #### Maven方式
-下载[最新版JAR包](https://search.maven.org/remotecontent?filepath=com/aliyun/openservices/tablestore/5.13.6/tablestore-5.13.6.jar)或者通过Maven：
+下载[最新版JAR包](https://search.maven.org/remotecontent?filepath=com/aliyun/openservices/tablestore/5.13.13/tablestore-5.13.13.jar)或者通过Maven：
 ```xml
 <dependency>
   <groupId>com.aliyun.openservices</groupId>
   <artifactId>tablestore</artifactId>
-  <version>5.13.12</version>
+  <version>5.13.13</version>
 </dependency>
 ```
 

pom.xml

@@ -3,7 +3,7 @@
   <modelVersion>4.0.0</modelVersion>  
   <groupId>com.aliyun.openservices</groupId>  
   <artifactId>tablestore</artifactId>  
-  <version>5.13.12</version>  
+  <version>5.13.13</version>  
   <packaging>jar</packaging>  
   <name>AliCloud TableStore SDK for Java</name>  
   <url>http://www.aliyun.com</url>  

src/main/java/com/alicloud/openservices/tablestore/core/protocol/timeseries/TimeseriesResponseFactory.java

@@ -288,12 +288,7 @@ public static ScanTimeseriesDataResponse createScanTimeseriesDataResponse(
                 List<PlainBufferRow> pbRows = inputStream.readRowsWithHeader();
                 List<TimeseriesRow> rows = new ArrayList<TimeseriesRow>(pbRows.size());
                 for (int i = 0; i < pbRows.size(); i++) {
-                    if (i > 0) {
-                        // reuse timeseries key
-                        rows.add(parseRowFromPlainbuffer(pbRows.get(i), rows.get(0).getTimeseriesKey()));
-                    } else {
-                        rows.add(parseRowFromPlainbuffer(pbRows.get(i)));
-                    }
+                    rows.add(parseRowFromPlainbuffer(pbRows.get(i)));
                 }
                 response.setRows(rows);
             } catch (IOException e) {

src/main/java/com/alicloud/openservices/tablestore/model/timeseries/ScanTimeseriesDataResponse.java

@@ -7,6 +7,8 @@
 
 public class ScanTimeseriesDataResponse extends Response {
 
+    public static String _VERSION_ = "20230111";
+
     private List<TimeseriesRow> rows = new ArrayList<TimeseriesRow>();
     private byte[] nextToken;
 

src/main/java/com/alicloud/openservices/tablestore/tunnel/pipeline/ProcessDataPipeline.java

@@ -145,7 +145,7 @@ public ProcessRecordsInput doProcess(ReadRecordsRequest readRecordsRequest) thro
                             } else {
                                 LOG.info("GetRecords, Num: {}, LoopTimes: {}, TotalBytes: {}, Channel connect: {}, Latency: {} ms, Next Token: {}",
                                         totalRecordsCount, times, totalBytes, connect, System.currentTimeMillis() - beginTs, resp.getNextToken());
-                                return new ProcessRecordsInput(totalRecords, resp.getNextToken(), resp.getRequestId());
+                                return new ProcessRecordsInput(totalRecords, resp.getNextToken(), resp.getRequestId(), connect.getChannelId());
                             }
                         } catch (Exception e) {
                             throw new StageException(this, readRecordsRequest, e.getMessage(), e);

src/main/java/com/alicloud/openservices/tablestore/tunnel/worker/ProcessRecordsInput.java

@@ -3,18 +3,30 @@
 import java.util.List;
 
 import com.alicloud.openservices.tablestore.model.StreamRecord;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class ProcessRecordsInput {
+    private static final Logger LOG = LoggerFactory.getLogger(ProcessRecordsInput.class);
+
     private List<StreamRecord> records;
     private String nextToken;
     private String traceId;
+    private String channelId;
 
     public ProcessRecordsInput(List<StreamRecord> records, String nextToken, String traceId) {
         this.records = records;
         this.nextToken = nextToken;
         this.traceId = traceId;
     }
 
+    public ProcessRecordsInput(List<StreamRecord> records, String nextToken, String traceId, String channelId) {
+        this.records = records;
+        this.nextToken = nextToken;
+        this.traceId = traceId;
+        this.channelId = channelId;
+    }
+
     public List<StreamRecord> getRecords() {
         return records;
     }
@@ -39,6 +51,18 @@ public void setTraceId(String traceId) {
         this.traceId = traceId;
     }
 
+    public String getChannelId() {
+        return channelId;
+    }
+
+    public String getPartitionId() {
+        String[] splits = channelId.split("_");
+        if (splits.length != 2) {
+            LOG.info("invalid channel id {}", channelId);
+        }
+        return splits[0];
+    }
+
     @Override
     public String toString() {
         //TODO

src/test/java/com/alicloud/openservices/tablestore/tunnel/functiontest/TestTunnelWorkerSimplePerf.java

@@ -30,13 +30,16 @@ public PerfElement(long timestamp, long speed, long totalCount) {
 
     private static final Gson GSON = new Gson();
     private static final int CAL_INTERVAL_MILLIS = 5000;
+
     static class PerfProcessor implements IChannelProcessor {
         private static final AtomicLong counter = new AtomicLong(0);
         private static final AtomicLong latestTs = new AtomicLong(0);
         private static final AtomicLong allCount = new AtomicLong(0);
 
         @Override
         public void process(ProcessRecordsInput input) {
+            System.out.println(input.getChannelId());
+            System.out.println(input.getPartitionId());
             counter.addAndGet(input.getRecords().size());
             allCount.addAndGet(input.getRecords().size());
             if (System.currentTimeMillis() - latestTs.get() > CAL_INTERVAL_MILLIS) {
@@ -62,7 +65,7 @@ public static void main(String[] args) {
         TunnelClient client = new TunnelClient(Endpoint, AccessId, AccessKey, InstanceName);
         TunnelWorkerConfig config = new TunnelWorkerConfig(new PerfProcessor());
         config.setHeartbeatIntervalInSec(15);
-        TunnelWorker worker = new TunnelWorker("265c97a9-2c41-4cb5-abdf-8b9966c4f0b8", client, config);
+        TunnelWorker worker = new TunnelWorker("0edc3bb4-ae62-4fe6-ae17-796d6801b476", client, config);
         try {
             worker.connectAndWorking();
         } catch (Exception e) {