Skip to content

Commit 23ca287

Browse files
maochongxinmaochongxin
authored
Merge pull request #161 from oceanbase/retry_with_errorcode
Refactor Retry Strategy: Adapt Retry Logic for Single Operations
2 parents 30fa10e + e5bcda1 commit 23ca287

File tree

7 files changed

+41
-24
lines changed

7 files changed

+41
-24
lines changed

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

Lines changed: 21 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -520,6 +520,7 @@ private <T> T execute(String tableName, TableExecuteCallback<T> callback, ObServ
520520
throw new IllegalArgumentException("table name is null");
521521
}
522522
boolean needRefreshTableEntry = false;
523+
boolean needFetchAllRouteInfo = false;
523524
int tryTimes = 0;
524525
long startExecute = System.currentTimeMillis();
525526
while (true) {
@@ -539,7 +540,7 @@ private <T> T execute(String tableName, TableExecuteCallback<T> callback, ObServ
539540
obPair = new ObPair<Long, ObTableParam>(0L, new ObTableParam(odpTable));
540541
} else {
541542
obPair = getTableBySingleRowKeyWithRoute(tableName, callback.getRowKey(),
542-
needRefreshTableEntry, tableEntryRefreshIntervalWait, route);
543+
needRefreshTableEntry, tableEntryRefreshIntervalWait, needFetchAllRouteInfo, route);
543544
}
544545
T t = callback.execute(obPair);
545546
resetExecuteContinuousFailureCount(tableName);
@@ -588,6 +589,11 @@ private <T> T execute(String tableName, TableExecuteCallback<T> callback, ObServ
588589
"retry while meet Exception needing refresh, errorCode: {} , errorMsg: {},retry times {}",
589590
((ObTableException) ex).getErrorCode(), ex.getMessage(),
590591
tryTimes);
592+
if (ex instanceof ObTableNeedFetchAllException) {
593+
needFetchAllRouteInfo = true;
594+
// reset failure count while fetch all route info
595+
this.resetExecuteContinuousFailureCount(tableName);
596+
}
591597
} else {
592598
calculateContinuousFailure(tableName, ex.getMessage());
593599
throw ex;
@@ -675,6 +681,7 @@ private <T> T executeMutation(String tableName, MutationExecuteCallback<T> callb
675681
throw new IllegalArgumentException("table name is null");
676682
}
677683
boolean needRefreshTableEntry = false;
684+
boolean needFetchAllRouteInfo = false;
678685
int tryTimes = 0;
679686
long startExecute = System.currentTimeMillis();
680687
while (true) {
@@ -696,12 +703,10 @@ private <T> T executeMutation(String tableName, MutationExecuteCallback<T> callb
696703
if (null != callback.getRowKey()) {
697704
// using row key
698705
obPair = getTableBySingleRowKeyWithRoute(tableName, callback.getRowKey(),
699-
needRefreshTableEntry, tableEntryRefreshIntervalWait, route);
706+
needRefreshTableEntry, tableEntryRefreshIntervalWait, needFetchAllRouteInfo, route);
700707
} else if (null != callback.getKeyRanges()) {
701708
// using scan range
702-
obPair = getTableByRowKeyRange(tableName, new ObTableQuery(),
703-
callback.getKeyRanges(), needRefreshTableEntry,
704-
tableEntryRefreshIntervalWait, route);
709+
obPair = getTableByRowKeyRange(tableName, new ObTableQuery(), callback.getKeyRanges());
705710
} else {
706711
throw new ObTableException("rowkey and scan range are null in mutation");
707712
}
@@ -754,6 +759,11 @@ private <T> T executeMutation(String tableName, MutationExecuteCallback<T> callb
754759
"retry while meet Exception needing refresh, errorCode: {} , errorMsg: {},retry times {}",
755760
((ObTableException) ex).getErrorCode(), ex.getMessage(),
756761
tryTimes);
762+
if (ex instanceof ObTableNeedFetchAllException) {
763+
needFetchAllRouteInfo = true;
764+
// reset failure count while fetch all route info
765+
this.resetExecuteContinuousFailureCount(tableName);
766+
}
757767
} else {
758768
calculateContinuousFailure(tableName, ex.getMessage());
759769
throw ex;
@@ -1429,14 +1439,15 @@ private ReplicaLocation getPartitionLocation(TableEntry tableEntry, long partId,
14291439
* @param rowKey row key
14301440
* @param refresh whether to refresh
14311441
* @param waitForRefresh whether wait for refresh
1442+
* @param needFetchAll whether need fetch all
14321443
* @return ObPair of partId and table
14331444
* @throws Exception exception
14341445
*/
14351446
public ObPair<Long, ObTableParam> getTableBySingleRowKey(String tableName, Object[] rowKey,
1436-
boolean refresh, boolean waitForRefresh)
1447+
boolean refresh, boolean waitForRefresh, boolean needFetchAll)
14371448
throws Exception {
14381449
ObServerRoute route = getRoute(false);
1439-
return getTableBySingleRowKeyWithRoute(tableName, rowKey, refresh, waitForRefresh, route);
1450+
return getTableBySingleRowKeyWithRoute(tableName, rowKey, refresh, waitForRefresh, needFetchAll, route);
14401451
}
14411452

14421453
/**
@@ -1453,9 +1464,10 @@ public ObPair<Long, ObTableParam> getTableBySingleRowKeyWithRoute(String tableNa
14531464
Object[] rowKey,
14541465
boolean refresh,
14551466
boolean waitForRefresh,
1467+
boolean needFetchAll,
14561468
ObServerRoute route)
14571469
throws Exception {
1458-
TableEntry tableEntry = getOrRefreshTableEntry(tableName, refresh, waitForRefresh, false);
1470+
TableEntry tableEntry = getOrRefreshTableEntry(tableName, refresh, waitForRefresh, needFetchAll);
14591471

14601472
long partId = getPartition(tableEntry, rowKey); // partition id in 3.x, origin partId in 4.x, logicId
14611473

@@ -1466,14 +1478,10 @@ public ObPair<Long, ObTableParam> getTableBySingleRowKeyWithRoute(String tableNa
14661478
* For mutation (queryWithFilter)
14671479
* @param tableName table want to get
14681480
* @param keyRanges key
1469-
* @param refresh whether to refresh
1470-
* @param waitForRefresh whether wait for refresh
1471-
* @param route ObServer route
14721481
* @return ObPair of partId and table
14731482
* @throws Exception exception
14741483
*/
1475-
public ObPair<Long, ObTableParam> getTableByRowKeyRange(String tableName, ObTableQuery query, List<ObNewRange> keyRanges, boolean refresh,
1476-
boolean waitForRefresh, ObServerRoute route)
1484+
public ObPair<Long, ObTableParam> getTableByRowKeyRange(String tableName, ObTableQuery query, List<ObNewRange> keyRanges)
14771485
throws Exception {
14781486
Map<Long, ObTableParam> partIdMapObTable = new HashMap<Long, ObTableParam>();
14791487
for (ObNewRange rang : keyRanges) {

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

Lines changed: 13 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -140,10 +140,18 @@ public ObPayload invokeSync(final ObTableConnection conn, final ObPayload reques
140140
}
141141
}
142142
if (resultCode.getRcode() != 0) {
143-
ExceptionUtil.throwObTableException(conn.getObTable().getIp(), conn.getObTable()
144-
.getPort(), response.getHeader().getTraceId1(), response.getHeader()
145-
.getTraceId0(), resultCode.getRcode(), resultCode.getErrMsg());
146-
return null;
143+
String errMessage = TraceUtil.formatTraceMessage(conn, request,
144+
"routed to the wrong server: " + response.getMessage());
145+
logger.warn(errMessage);
146+
if (needFetchAll(resultCode.getRcode(), resultCode.getPcode())) {
147+
throw new ObTableNeedFetchAllException(errMessage);
148+
} else if (needFetchPartial(resultCode.getRcode())) {
149+
throw new ObTableRoutingWrongException(errMessage);
150+
} else {
151+
ExceptionUtil.throwObTableException(conn.getObTable().getIp(), conn.getObTable()
152+
.getPort(), response.getHeader().getTraceId1(), response.getHeader()
153+
.getTraceId0(), resultCode.getRcode(), resultCode.getErrMsg());
154+
}
147155
}
148156

149157
// decode payload itself
@@ -178,6 +186,7 @@ protected InvokeFuture createInvokeFuture(Connection conn, RemotingCommand reque
178186
InvokeCallback invokeCallback) {
179187
return new ObClientFuture(request.getId());
180188
}
189+
181190

182191
// schema changed
183192
private boolean needFetchAll(int errorCode, int pcode) {

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -244,7 +244,7 @@ public Map<Long, ObPair<ObTableParam, List<ObPair<Integer, ObTableOperation>>>>
244244
rowKey[j] = rowKeyObject.getObj(j).getValue();
245245
}
246246
ObPair<Long, ObTableParam> tableObPair = obTableClient.getTableBySingleRowKeyWithRoute(
247-
tableName, rowKey, false, false,
247+
tableName, rowKey, false, false, false,
248248
obTableClient.getRoute(batchOperation.isReadOnly()));
249249
ObPair<ObTableParam, List<ObPair<Integer, ObTableOperation>>> obTableOperations = partitionOperationsMap
250250
.get(tableObPair.getLeft());

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -340,7 +340,7 @@ public Map<Long, Map<Long, ObPair<ObTableParam, List<ObPair<Integer, ObTableSing
340340
rowKey[j] = rowkeyObjs.get(j).getValue();
341341
}
342342
ObPair<Long, ObTableParam> tableObPair= obTableClient.getTableBySingleRowKeyWithRoute(tableName, rowKey,
343-
false, false, obTableClient.getRoute(false));
343+
false, false, false, obTableClient.getRoute(false));
344344
long lsId = tableObPair.getRight().getLsId();
345345

346346
Map<Long, ObPair<ObTableParam, List<ObPair<Integer, ObTableSingleOp>>>> tabletOperations

‎src/test/java/com/alipay/oceanbase/rpc/bolt/ObTableTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -48,7 +48,7 @@ public void setup() throws Exception {
4848
throw new ObTableException("ODP Mode does not support this test");
4949
} else {
5050
obTable = obTableClient
51-
.getTableBySingleRowKey("test_varchar_table", new Object[] { "abc" }, true, true).getRight()
51+
.getTableBySingleRowKey("test_varchar_table", new Object[] { "abc" }, true, true, false).getRight()
5252
.getObTable();
5353
client = obTable;
5454
}

‎src/test/java/com/alipay/oceanbase/rpc/hbase/ObHTableTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -55,7 +55,7 @@ public void setup() throws Exception {
5555
throw new ObTableException("ODP Mode does not support this test");
5656
} else {
5757
client = obTableClient
58-
.getTableBySingleRowKey("test_varchar_table", new Object[] { "abc" }, true, true).getRight()
58+
.getTableBySingleRowKey("test_varchar_table", new Object[] { "abc" }, true, true, false).getRight()
5959
.getObTable();
6060
this.obTableClient = obTableClient;
6161
}

‎src/test/java/com/alipay/oceanbase/rpc/table/ObTableConnectionTest.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -77,7 +77,7 @@ public void testConnectionPoolSize() throws Exception {
7777
.getObTableConnectionPoolSize());
7878
} else {
7979
ObPair<Long, ObTableParam> obPair = obTableClient.getTableBySingleRowKey("test_varchar_table",
80-
new String[] { "abc" }, false, false);
80+
new String[] { "abc" }, false, false, false);
8181
int poolSize = obPair.getRight().getObTable().getObTableConnectionPoolSize();
8282
assertEquals(TEST_CONNECTION_POOL_SIZE, poolSize);
8383
}
@@ -109,7 +109,7 @@ public void testWatermarkSetting() throws Exception {
109109
.getNettyBlockingWaitInterval());
110110
} else {
111111
ObPair<Long, ObTableParam> obPair = obTableClient.getTableBySingleRowKey("test_varchar_table",
112-
new String[] { "abc" }, false, false);
112+
new String[] { "abc" }, false, false, false);
113113
int lowWatermark = obPair.getRight().getObTable().getNettyBufferLowWatermark();
114114
int highWatermark = obPair.getRight().getObTable().getNettyBufferHighWatermark();
115115
int waitInterval = obPair.getRight().getObTable().getNettyBlockingWaitInterval();
@@ -131,7 +131,7 @@ public void testDefaultWatermark() throws Exception {
131131
// do nothing
132132
} else {
133133
ObPair<Long, ObTableParam> obPair = obTableClient.getTableBySingleRowKey("test_varchar_table",
134-
new String[] { "abc" }, false, false);
134+
new String[] { "abc" }, false, false, false);
135135
int lowWatermark = obPair.getRight().getObTable().getNettyBufferLowWatermark();
136136
int highWatermark = obPair.getRight().getObTable().getNettyBufferHighWatermark();
137137
int waitInterval = obPair.getRight().getObTable().getNettyBlockingWaitInterval();

0 commit comments

Comments
 (0)