add retry logic for common query

Author: maochongxin

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

@@ -45,6 +45,8 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.ReentrantLock;
 
+import static com.alipay.oceanbase.rpc.util.TableClientLoggerFactory.RUNTIME;
+
 public abstract class AbstractQueryStreamResult extends AbstractPayload implements
                                                                        QueryStreamResult {
 
@@ -61,14 +63,15 @@ public abstract class AbstractQueryStreamResult extends AbstractPayload implemen
     // global index: key is index table name (be like: __idx_<data_table_id>_<index_name>)
     protected String                                                           indexTableName;
     protected ObTableEntityType                                                entityType;
-    protected Map<Long, ObPair<Long, ObTableParam>>                            expectant;                                                                                     // Map<logicId, ObPair<logicId, param>>
+    protected Map<Long, ObPair<Long, ObTableParam>>                            expectant;
     protected List<String>                                                     cacheProperties     = new LinkedList<String>();
     protected LinkedList<List<ObObj>>                                          cacheRows           = new LinkedList<List<ObObj>>();
     private LinkedList<ObPair<ObPair<Long, ObTableParam>, ObTableQueryResult>> partitionLastResult = new LinkedList<ObPair<ObPair<Long, ObTableParam>, ObTableQueryResult>>();
     private ObReadConsistency                                                  readConsistency     = ObReadConsistency.STRONG;
     // ObRowKey objs: [startKey, MIN_OBJECT, MIN_OBJECT]
     public List<ObObj>                                                         currentStartKey;
-
+    protected ObTableClient          client;
+    
     /*
      * Get pcode.
      */
@@ -549,9 +552,32 @@ public void init() throws Exception {
             return;
         }
         if (tableQuery.getBatchSize() == -1) {
-            for (Map.Entry<Long, ObPair<Long, ObTableParam>> entry : expectant.entrySet()) {
-                // mark the refer partition
-                referToNewPartition(entry.getValue());
+            if (!expectant.isEmpty()) {
+                Iterator<Map.Entry<Long, ObPair<Long, ObTableParam>>> it = expectant.entrySet()
+                        .iterator();
+                int retryTimes = 0;
+                while (it.hasNext()) {
+                    Map.Entry<Long, ObPair<Long, ObTableParam>> entry = it.next();
+                    try {
+                        // try access new partition, async will not remove useless expectant
+                        referToNewPartition(entry.getValue());
+                    } catch (Exception e) {
+                        if (e instanceof ObTableNeedFetchAllException) {
+                            setExpectant(refreshPartition(tableQuery, tableName));
+                            it = expectant.entrySet().iterator();
+                            retryTimes++;
+                            if (retryTimes > client.getRuntimeRetryTimes()) {
+                                RUNTIME.error("Fail to get refresh table entry response after {}",
+                                        retryTimes);
+                                throw new ObTableRetryExhaustedException(
+                                        "Fail to get refresh table entry response after " + retryTimes);
+
+                            }
+                        } else {
+                            throw e;
+                        }
+                    }
+                }
             }
             expectant.clear();
         } else {
@@ -692,4 +718,19 @@ public ObReadConsistency getReadConsistency() {
     public void setReadConsistency(ObReadConsistency readConsistency) {
         this.readConsistency = readConsistency;
     }
+
+    /**
+     * Get client.
+     * @return client
+     */
+    public ObTableClient getClient() {
+        return client;
+    }
+
+    /*
+     * Set client.
+     */
+    public void setClient(ObTableClient client) {
+        this.client = client;
+    }
 }

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

@@ -44,7 +44,6 @@
 public class ObTableClientQueryAsyncStreamResult extends AbstractQueryStreamResult {
     private static final Logger      logger         = LoggerFactory
                                                         .getLogger(ObTableClientQueryStreamResult.class);
-    protected ObTableClient          client;
     private boolean                  isEnd          = true;
     private long                     sessionId      = Constants.OB_INVALID_ID;
     private ObTableQueryAsyncRequest asyncRequest   = new ObTableQueryAsyncRequest();
@@ -338,19 +337,7 @@ public void close() throws Exception {
             closeLastStreamResult(lastEntry.getValue());
         }
     }
-
-    public ObTableClient getClient() {
-        return client;
-    }
-
-    /**
-     * Set client.
-     * @param client client want to set
-     */
-    public void setClient(ObTableClient client) {
-        this.client = client;
-    }
-
+    
     public boolean isEnd() {
         return isEnd;
     }

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

@@ -34,10 +34,8 @@
 import java.util.concurrent.atomic.AtomicReference;
 
 public class ObTableClientQueryStreamResult extends AbstractQueryStreamResult {
-
     private static final Logger logger = TableClientLoggerFactory
                                            .getLogger(ObTableClientQueryStreamResult.class);
-    protected ObTableClient     client;
 
     protected ObTableQueryResult referToNewPartition(ObPair<Long, ObTableParam> partIdWithObTable)
                                                                                                   throws Exception {
@@ -84,19 +82,4 @@ protected Map<Long, ObPair<Long, ObTableParam>> refreshPartition(ObTableQuery ta
                                                                                       throws Exception {
         return buildPartitions(client, tableQuery, tableName);
     }
-
-    /**
-     * Get client.
-     * @return client
-     */
-    public ObTableClient getClient() {
-        return client;
-    }
-
-    /*
-     * Set client.
-     */
-    public void setClient(ObTableClient client) {
-        this.client = client;
-    }
 }

src/main/java/com/alipay/oceanbase/rpc/table/ObTableClientLSBatchOpsImpl.java

@@ -631,7 +631,7 @@ private void executeWithRetries(
         Map<Long, Map<Long, ObPair<ObTableParam, List<ObPair<Integer, ObTableSingleOp>>>>> currentPartitions = new HashMap<>();
         currentPartitions.put(entry.getKey(), entry.getValue());
 
-        while (retryCount < maxRetries && !success) {
+        while (retryCount <= maxRetries && !success) {
             boolean allPartitionsSuccess = true;
 
             for (Map.Entry<Long, Map<Long, ObPair<ObTableParam, List<ObPair<Integer, ObTableSingleOp>>>>> currentEntry : currentPartitions.entrySet()) {