add retry logic for ObTableQueryAndMutateRequest

Author: maochongxin

src/main/java/com/alipay/oceanbase/rpc/ObTableClient.java

@@ -3133,41 +3133,95 @@ public ObPayload execute(final ObTableAbstractOperationRequest request) throws E
                 request.setTimeout(getOdpTable().getObTableOperationTimeout());
                 return getOdpTable().execute(request);
             } else {
+                int maxRetries = getRuntimeRetryTimes(); // Define the maximum number of retries  
+                int tryTimes = 0;
+                long startExecute = System.currentTimeMillis();
+                boolean needRefreshTableEntry = false;
                 Map<Long, ObTableParam> partIdMapObTable = new HashMap<Long, ObTableParam>();
-                for (ObNewRange rang : tableQuery.getKeyRanges()) {
-                    ObRowKey startKey = rang.getStartKey();
-                    int startKeySize = startKey.getObjs().size();
-                    ObRowKey endKey = rang.getEndKey();
-                    int endKeySize = endKey.getObjs().size();
-                    Object[] start = new Object[startKeySize];
-                    Object[] end = new Object[endKeySize];
-                    for (int i = 0; i < startKeySize; i++) {
-                        start[i] = startKey.getObj(i).getValue();
+                while (true) {
+                    long currentExecute = System.currentTimeMillis();
+                    long costMillis = currentExecute - startExecute;
+                    if (costMillis > getRuntimeMaxWait()) {
+                        logger.error(
+                                "tablename:{} it has tried " + tryTimes
+                                        + " times and it has waited " + costMillis
+                                        + "/ms which exceeds response timeout "
+                                        + getRuntimeMaxWait() + "/ms", request.getTableName());
+                        throw new ObTableTimeoutExcetion("it has tried " + tryTimes
+                                + " times and it has waited " + costMillis
+                                + "/ms which exceeds response timeout "
+                                + getRuntimeMaxWait() + "/ms");
                     }
+                    try {
+                        // Recalculate partIdMapObTable
+                        // Clear the map before recalculating
+                        partIdMapObTable.clear(); 
+                        for (ObNewRange rang : tableQuery.getKeyRanges()) {
+                            ObRowKey startKey = rang.getStartKey();
+                            int startKeySize = startKey.getObjs().size();
+                            ObRowKey endKey = rang.getEndKey();
+                            int endKeySize = endKey.getObjs().size();
+                            Object[] start = new Object[startKeySize];
+                            Object[] end = new Object[endKeySize];
+                            for (int i = 0; i < startKeySize; i++) {
+                                start[i] = startKey.getObj(i).getValue();
+                            }
 
-                    for (int i = 0; i < endKeySize; i++) {
-                        end[i] = endKey.getObj(i).getValue();
-                    }
-                    ObBorderFlag borderFlag = rang.getBorderFlag();
-                    List<ObPair<Long, ObTableParam>> pairList = getTables(request.getTableName(),
-                        tableQuery, start, borderFlag.isInclusiveStart(), end,
-                        borderFlag.isInclusiveEnd(), false, false);
-                    for (ObPair<Long, ObTableParam> pair : pairList) {
-                        partIdMapObTable.put(pair.getLeft(), pair.getRight());
-                    }
-                }
-                if (partIdMapObTable.size() > 1) {
-                    throw new ObTablePartitionConsistentException(
-                        "query and mutate must be a atomic operation");
-                }
+                            for (int i = 0; i < endKeySize; i++) {
+                                end[i] = endKey.getObj(i).getValue();
+                            }
+                            ObBorderFlag borderFlag = rang.getBorderFlag();
+                            List<ObPair<Long, ObTableParam>> pairList = getTables(request.getTableName(),
+                                    tableQuery, start, borderFlag.isInclusiveStart(), end,
+                                    borderFlag.isInclusiveEnd(), needRefreshTableEntry, isTableEntryRefreshIntervalWait());
+                            for (ObPair<Long, ObTableParam> pair : pairList) {
+                                partIdMapObTable.put(pair.getLeft(), pair.getRight());
+                            }
+                        }
 
-                for (Long partId : partIdMapObTable.keySet()) {
-                    ObTableParam tableParam = partIdMapObTable.get(partId);
-                    request.setTableId(tableParam.getTableId());
-                    request.setPartitionId(tableParam.getPartitionId());
-                    request.setTimeout(tableParam.getObTable().getObTableOperationTimeout());
-                    ObTable obTable = tableParam.getObTable();
-                    return executeWithRetry(obTable, request, request.getTableName());
+                        // Check if partIdMapObTable size is greater than 1  
+                        if (partIdMapObTable.size() > 1) {
+                            throw new ObTablePartitionConsistentException(
+                                    "query and mutate must be a atomic operation");
+                        }
+                        // Proceed with the operation
+                        Map.Entry<Long, ObTableParam> entry = partIdMapObTable.entrySet().iterator().next();
+                        ObTableParam tableParam = entry.getValue();
+                        request.setTableId(tableParam.getTableId());
+                        request.setPartitionId(tableParam.getPartitionId());
+                        request.setTimeout(tableParam.getObTable().getObTableOperationTimeout());
+                        ObTable obTable = tableParam.getObTable();
+
+                        // Attempt to execute the operation  
+                        return executeWithRetry(obTable, request, request.getTableName());
+                    } catch (Exception ex) {
+                        tryTimes++;
+                        if (ex instanceof ObTableException && ((ObTableException) ex).isNeedRefreshTableEntry()) {
+                            needRefreshTableEntry = true;
+                            logger.warn(
+                                    "tablename:{} partition id:{} batch ops refresh table while meet ObTableMasterChangeException, errorCode: {}",
+                                    request.getTableName(), request.getPartitionId(), ((ObTableException) ex).getErrorCode(), ex);
+
+                            if (isRetryOnChangeMasterTimes() && tryTimes < maxRetries) {
+                                logger.warn(
+                                        "tablename:{} partition id:{} batch ops retry while meet ObTableMasterChangeException, errorCode: {} , retry times {}",
+                                        request.getTableName(), request.getPartitionId(), ((ObTableException) ex).getErrorCode(),
+                                        tryTimes, ex);
+
+                                if (ex instanceof ObTableNeedFetchAllException) {
+                                    // Refresh table info  
+                                    getOrRefreshTableEntry(request.getTableName(), needRefreshTableEntry, isTableEntryRefreshIntervalWait(), true);
+                                }
+                            } else {
+                                calculateContinuousFailure(request.getTableName(), ex.getMessage());
+                                throw ex;
+                            }
+                        } else {
+                            calculateContinuousFailure(request.getTableName(), ex.getMessage());
+                            // Handle other exceptions or rethrow
+                            throw ex;
+                        }
+                    }
                 }
             }
         }