fix: truncate retry fail

Author: maochongxin

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

@@ -128,7 +128,7 @@ public class ObTableClient extends AbstractObTableClient implements Lifecycle {
     /*
      * TableName -> Failures/Lock
      */
-    private ConcurrentHashMap<String, AtomicLong>             tableContinuousFailures                 = new ConcurrentHashMap<String, AtomicLong>();
+    public ConcurrentHashMap<String, AtomicLong>             tableContinuousFailures                 = new ConcurrentHashMap<String, AtomicLong>();
 
     private ConcurrentHashMap<String, Lock>                   refreshTableLocks                       = new ConcurrentHashMap<String, Lock>();
 

src/main/java/com/alipay/oceanbase/rpc/bolt/transport/ObTableRemoting.java

@@ -21,10 +21,7 @@
 import com.alipay.oceanbase.rpc.bolt.protocol.ObTablePacketCode;
 import com.alipay.oceanbase.rpc.exception.*;
 import com.alipay.oceanbase.rpc.protocol.packet.ObCompressType;
-import com.alipay.oceanbase.rpc.protocol.payload.AbstractPayload;
-import com.alipay.oceanbase.rpc.protocol.payload.Credentialable;
-import com.alipay.oceanbase.rpc.protocol.payload.ObPayload;
-import com.alipay.oceanbase.rpc.protocol.payload.ObRpcResultCode;
+import com.alipay.oceanbase.rpc.protocol.payload.*;
 import com.alipay.oceanbase.rpc.protocol.payload.impl.login.ObTableLoginRequest;
 import com.alipay.oceanbase.rpc.util.ObPureCrc32C;
 import com.alipay.oceanbase.rpc.util.TableClientLoggerFactory;
@@ -108,14 +105,6 @@ public ObPayload invokeSync(final ObTableConnection conn, final ObPayload reques
                 throw new FeatureNotSupportedException(errMessage);
             }
             ByteBuf buf = response.getPacketContentBuf();
-            // If response indicates the request is routed to wrong server, we should refresh the routing meta.
-            if (response.getHeader().isRoutingWrong()) {
-                String errMessage = TraceUtil.formatTraceMessage(conn, request,
-                    "routed to the wrong server: " + response.getMessage());
-                logger.warn(errMessage);
-                throw new ObTableRoutingWrongException(errMessage);
-            }
-
             // verify checksum
             long expected_checksum = response.getHeader().getChecksum();
             byte[] content = new byte[buf.readableBytes()];
@@ -132,7 +121,17 @@ public ObPayload invokeSync(final ObTableConnection conn, final ObPayload reques
             // decode ResultCode for response packet
             ObRpcResultCode resultCode = new ObRpcResultCode();
             resultCode.decode(buf);
-
+            // If response indicates the request is routed to wrong server, we should refresh the routing meta.
+            if (response.getHeader().isRoutingWrong()) {
+                String errMessage = TraceUtil.formatTraceMessage(conn, request,
+                        "routed to the wrong server: " + response.getMessage());
+                logger.warn(errMessage);
+                if (needFetchAll(resultCode.getRcode())) {
+                    throw new ObTableRoutingWrongException(errMessage);
+                } else {
+                    throw new ObTableMasterChangeException(errMessage);
+                }
+            }
             if (resultCode.getRcode() != 0) {
                 ExceptionUtil.throwObTableException(conn.getObTable().getIp(), conn.getObTable()
                     .getPort(), response.getHeader().getTraceId1(), response.getHeader()
@@ -173,4 +172,11 @@ protected InvokeFuture createInvokeFuture(Connection conn, RemotingCommand reque
         return new ObClientFuture(request.getId());
     }
 
+    private boolean needFetchAll(int errorCode) {
+        return errorCode == ResultCodes.OB_PARTITION_NOT_EXIST.errorCode
+                || errorCode == ResultCodes.OB_INVALID_PARTITION.errorCode
+                || errorCode == ResultCodes.OB_UNKNOWN_PARTITION.errorCode
+                || errorCode == ResultCodes.OB_PARTITION_NOT_MATCH.errorCode
+                || errorCode == ResultCodes.OB_TABLET_NOT_EXIST.errorCode;
+    }
 }

src/main/java/com/alipay/oceanbase/rpc/protocol/payload/ResultCodes.java

@@ -358,6 +358,7 @@ public enum ResultCodes {
     OB_CLUSTER_NO_MATCH(-4666), //
     OB_CHECK_ZONE_MERGE_ORDER(-4667), //
     OB_ERR_ZONE_NOT_EMPTY(-4668), //
+    OB_TABLET_NOT_EXIST(-4725), //
     OB_ERR_PARSER_INIT(-5000), //
     OB_ERR_PARSE_SQL(-5001), //
     OB_ERR_RESOLVE_SQL(-5002), //

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

@@ -332,10 +332,12 @@ public void partitionExecute(ObTableOperationResult[] results,
                         if (failedServerList != null) {
                             route.setBlackList(failedServerList);
                         }
-                        subObTable = obTableClient
+                        ObTableParam newParam = obTableClient
                             .getTable(tableName, originPartId, needRefreshTableEntry,
-                                obTableClient.isTableEntryRefreshIntervalWait(), route).getRight()
-                            .getObTable();
+                                        obTableClient.isTableEntryRefreshIntervalWait(), route).getRight();
+
+                        subObTable = newParam.getObTable();
+                        subRequest.setPartitionId(newParam.getPartitionId());
                     }
                 }
                 subObTableBatchOperationResult = (ObTableBatchOperationResult) subObTable
@@ -380,6 +382,14 @@ public void partitionExecute(ObTableOperationResult[] results,
                                 "tablename:{} partition id:{} batch ops retry while meet ObTableMasterChangeException, errorCode: {} , retry times {}",
                                 tableName, partId, ((ObTableException) ex).getErrorCode(),
                                 tryTimes, ex);
+                        if (ex instanceof ObTableRoutingWrongException) {
+                            System.out.println("need fetchAll refresh");
+                            obTableClient.getOrRefreshTableEntry(tableName, true, obTableClient.isTableEntryRefreshIntervalWait(), true);
+                            obTableClient.tableContinuousFailures.computeIfPresent(tableName, (k, v) -> {
+                                v.set(0);
+                                return v;
+                            });
+                        }
                     } else {
                         obTableClient.calculateContinuousFailure(tableName, ex.getMessage());
                         throw ex;