support hbase scan renewLease (#211)

Author: stuBirdFly

src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/query/AbstractQueryStreamResult.java

@@ -284,6 +284,13 @@ protected ObPayload commonExecute(ObTableClient client, Logger logger,
         return result;
     }
 
+    /*
+     * RenewLease.
+     */
+    public void renewLease() throws Exception {
+        throw new IllegalStateException("renew only support stream query");
+    }
+
     /*
      * Next.
      */

src/main/java/com/alipay/oceanbase/rpc/protocol/payload/impl/execute/syncquery/ObQueryOperationType.java

@@ -21,7 +21,7 @@
 import java.util.Map;
 
 public enum ObQueryOperationType {
-    QUERY_START(0), QUERY_NEXT(1), QUERY_END(2);
+    QUERY_START(0), QUERY_NEXT(1), QUERY_END(2), QUERY_RENEW(3);
 
     private int                                       value;
     private static Map<Integer, ObQueryOperationType> map = new HashMap<Integer, ObQueryOperationType>();

src/main/java/com/alipay/oceanbase/rpc/stream/ObTableClientQueryAsyncStreamResult.java

@@ -181,6 +181,32 @@ protected Map<Long, ObPair<Long, ObTableParam>> refreshPartition(ObTableQuery ta
         return buildPartitions(client, tableQuery, tableName);
     }
 
+    // This function is designed for HBase-type requests.
+    // It is used to extend the session duration of a scan
+    @Override
+    public void renewLease() throws Exception {
+        if (!isEnd() && !expectant.isEmpty()) {
+            Iterator<Map.Entry<Long, ObPair<Long, ObTableParam>>> it = expectant.entrySet()
+                    .iterator();
+            Map.Entry<Long, ObPair<Long, ObTableParam>> lastEntry = it.next();
+            ObPair<Long, ObTableParam> partIdWithObTable = lastEntry.getValue();
+            // try access new partition, async will not remove useless expectant
+            ObTableParam obTableParam = partIdWithObTable.getRight();
+            ObTableQueryRequest queryRequest = asyncRequest.getObTableQueryRequest();
+
+            // refresh request info
+            queryRequest.setPartitionId(obTableParam.getPartitionId());
+            queryRequest.setTableId(obTableParam.getTableId());
+
+            // refresh async query request
+            asyncRequest.setQueryType(ObQueryOperationType.QUERY_RENEW);
+            asyncRequest.setQuerySessionId(sessionId);
+            executeAsync(partIdWithObTable, asyncRequest);
+        } else {
+            throw new ObTableException("query end or expectant is null");
+        }
+    }
+
     @Override
     public boolean next() throws Exception {
         checkStatus();