fix log && fix getPartitionReplica return wrong part id

Author: maochongxin

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

@@ -885,8 +885,9 @@ private <T> T execute(String tableName, OperationExecuteCallback<T> callback,
                             }
                         } else {
                             String logMessage = String.format(
-                                    "exhaust retry while meet NeedRefresh Exception, table name: %s, batch ops refresh table, errorCode: %d",
+                                    "exhaust retry while meet NeedRefresh Exception, table name: %s, retry times: %d, errorCode: %d",
                                     tableName,
+                                    runtimeRetryTimes,
                                     ((ObTableException) ex).getErrorCode()
                             );
                             logger.warn(logMessage, ex);
@@ -2041,12 +2042,10 @@ public ObPair<Long, ObTableParam> getTableInternal(String tableName, TableEntry
         ObTableParam param = createTableParam(obTable, tableEntry, obPartitionLocationInfo, partId, tabletId);
         if (ObGlobal.obVsnMajor() >= 4) {
         } else {
-            param.setPartId(partId);
-            param.setTableId(tableEntry.getTableId());
             param.setPartitionId(partId);
         }
         addr.recordAccess();
-        return new ObPair<>(tabletId, param);
+        return new ObPair<>(partId, param);
     }
 
     private ObPartitionLocationInfo getOrRefreshPartitionInfo(TableEntry tableEntry,
@@ -2116,7 +2115,7 @@ private List<ObPair<Long, ReplicaLocation>> getPartitionReplica(TableEntry table
             if (ObGlobal.obVsnMajor() >= 4) {
                 long tabletId = getTabletIdByPartId(tableEntry, 0L);
                 ObPartitionLocationInfo locationInfo = getOrRefreshPartitionInfo(tableEntry, tableName, tabletId);
-                replicas.add(new ObPair<>(tabletId, getPartitionLocation(locationInfo, route)));
+                replicas.add(new ObPair<>(0L, getPartitionLocation(locationInfo, route)));
             } else {
                 replicas.add(new ObPair<Long, ReplicaLocation>(0L, getPartitionLocation(tableEntry, 0L,
                         route)));
@@ -2131,7 +2130,7 @@ private List<ObPair<Long, ReplicaLocation>> getPartitionReplica(TableEntry table
             for (Long partId : partIds) {
                 long tabletId = getTabletIdByPartId(tableEntry, partId);
                 ObPartitionLocationInfo locationInfo = getOrRefreshPartitionInfo(tableEntry, tableName, tabletId);
-                replicas.add(new ObPair<>(tabletId, getPartitionLocation(locationInfo, route)));
+                replicas.add(new ObPair<>(partId, getPartitionLocation(locationInfo, route)));
             }
         } else {
             for (Long partId : partIds) {

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

@@ -165,7 +165,7 @@ public ObPayload invokeSync(final ObTableConnection conn, final ObPayload reques
             } else {
                 String errMessage = TraceUtil.formatTraceMessage(conn, response,
                     "receive unexpected command code: " + response.getCmdCode().value());
-                throw new ObTableUnexpectedException(errMessage);
+                throw new ObTableUnexpectedException(errMessage, resultCode.getRcode());
             }
 
             payload.decode(buf);

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

@@ -259,26 +259,22 @@ protected ObPayload commonExecute(ObTableClient client, Logger logger,
                         }
                         if (((ObTableException) e).isNeedRefreshTableEntry()) {
                             needRefreshTableEntry = true;
-                            logger
-                                .warn(
-                                    "tablename:{} partition id:{} stream query refresh table while meet Exception needing refresh, errorCode: {}",
-                                    indexTableName, partIdWithIndex.getLeft(),
-                                    ((ObTableException) e).getErrorCode(), e);
                             if (client.isRetryOnChangeMasterTimes()
                                 && (tryTimes - 1) < client.getRuntimeRetryTimes()) {
-                                logger
-                                    .warn(
-                                        "tablename:{} partition id:{} stream query retry while meet Exception needing refresh, errorCode: {} , retry times {}",
-                                        indexTableName, partIdWithIndex.getLeft(),
-                                        ((ObTableException) e).getErrorCode(), tryTimes, e);
                                 // tablet not exists, refresh table entry
                                 if (e instanceof ObTableNeedFetchAllException) {
                                     client.getOrRefreshTableEntry(indexTableName, true, true, true);
                                     throw e;
                                 }
                             } else {
+                                String logMessage = String.format(
+                                        "exhaust retry while meet NeedRefresh Exception, table name: %s, ls id: %d, batch ops refresh table, errorCode: %d",
+                                        indexTableName,
+                                        ((ObTableException) e).getErrorCode()
+                                );
+                                logger.warn(logMessage, e);
                                 client.calculateContinuousFailure(indexTableName, e.getMessage());
-                                throw e;
+                                throw new ObTableRetryExhaustedException(logMessage, e);
                             }
                         } else {
                             client.calculateContinuousFailure(indexTableName, e.getMessage());

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

@@ -432,24 +432,24 @@ public void partitionExecute(ObTableOperationResult[] results,
                 } else if (ex instanceof ObTableException
                            && ((ObTableException) ex).isNeedRefreshTableEntry()) {
                     needRefreshTableEntry = true;
-                    logger
-                        .warn(
-                            "tablename:{} partition id:{} batch ops refresh table while meet ObTableMasterChangeException, errorCode: {}",
-                            tableName, partId, ((ObTableException) ex).getErrorCode(), ex);
                     if (obTableClient.isRetryOnChangeMasterTimes()
                         && (tryTimes - 1) < obTableClient.getRuntimeRetryTimes()) {
                         if (ex instanceof ObTableNeedFetchAllException) {
-                            logger.warn("tablename:{}, partition_id: {}  batch ops retry while meet ObTableNeedFetchAllException, errorCode: {} , retry times {}",
-                                    tableName, subRequest.getPartitionId(),((ObTableException) ex).getErrorCode(),
-                                    tryTimes, ex);
                             // refresh table info
                             obTableClient.getOrRefreshTableEntry(tableName, needRefreshTableEntry,
                                 obTableClient.isTableEntryRefreshIntervalWait(), true);
                             throw ex;
                         }
                     } else {
+                        String logMessage = String.format(
+                                "exhaust retry while meet NeedRefresh Exception, table name: %s, batch ops refresh table, retry times: %d, errorCode: %d",
+                                tableName,
+                                obTableClient.getRuntimeRetryTimes(),
+                                ((ObTableException) ex).getErrorCode()
+                        );
+                        logger.warn(logMessage, ex);
                         obTableClient.calculateContinuousFailure(tableName, ex.getMessage());
-                        throw ex;
+                        throw new ObTableRetryExhaustedException(logMessage, ex);
                     }
                 } else {
                     obTableClient.calculateContinuousFailure(tableName, ex.getMessage());

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

@@ -519,6 +519,9 @@ public void partitionExecute(ObTableSingleOpResult[] results,
                                 false, false);
                         if (ObGlobal.obVsnMajor() >= 4) {
                             obTableClient.refreshTableLocationByTabletId(entry, realTableName, obTableClient.getTabletIdByPartId(entry, originPartId));
+                        } else { // 3.x
+                            obTableClient.getOrRefreshTableEntry(realTableName, needRefreshTableEntry,
+                                    obTableClient.isTableEntryRefreshIntervalWait(), false);
                         }
                         subObTable = obTableClient.getTableWithPartId(realTableName, originPartId, needRefreshTableEntry,
                                         obTableClient.isTableEntryRefreshIntervalWait(), false, route).
@@ -571,9 +574,10 @@ public void partitionExecute(ObTableSingleOpResult[] results,
                         }
                     } else {
                         String logMessage = String.format(
-                                "exhaust retry while meet NeedRefresh Exception, table name: %s, ls id: %d, batch ops refresh table, errorCode: %d",
+                                "exhaust retry while meet NeedRefresh Exception, table name: %s, ls id: %d, batch ops refresh table, retry times: %d, errorCode: %d",
                                 realTableName,
                                 lsId,
+                                obTableClient.getRuntimeRetryTimes(),
                                 ((ObTableException) ex).getErrorCode()
                         );
                         logger.warn(logMessage, ex);