Skip to content

Commit 041bb5d

Browse files
JackShi148JackShi148
committed
refresh table meta and locatino based on flag, refresh cache after retry successfully in server
1 parent fb6aa87 commit 041bb5d

File tree

8 files changed

+112
-44
lines changed

8 files changed

+112
-44
lines changed

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

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -837,7 +837,7 @@ public void calculateContinuousFailure(String tableName, String errorMsg) throws
837837
if (failures.incrementAndGet() > runtimeContinuousFailureCeiling) {
838838
logger.warn("refresh table entry {} while execute failed times exceeded {}, msg: {}",
839839
tableName, runtimeContinuousFailureCeiling, errorMsg);
840-
refreshMeta(tableName);
840+
tableRoute.refreshMeta(tableName);
841841
failures.set(0);
842842
} else {
843843
logger.warn("error msg: {}, current continues failure count: {}", errorMsg, failures);
@@ -968,14 +968,6 @@ public TableEntry getOrRefreshTableEntry(final String tableName, boolean forceRe
968968
if (!forceRefresh) {
969969
return tableRoute.getTableEntry(tableName);
970970
}
971-
return refreshMeta(tableName);
972-
}
973-
974-
/**
975-
* refresh table meta information except location
976-
* @param tableName table name
977-
* */
978-
private TableEntry refreshMeta(String tableName) throws Exception {
979971
return tableRoute.refreshMeta(tableName);
980972
}
981973

@@ -2102,6 +2094,14 @@ public ObPayload executeWithRetry(ObTable obTable, ObPayload request, String tab
21022094
"Rerouting return IP is {}", moveResponse.getReplica().getServer().ipToString(), move .getReplica().getServer().ipToString());
21032095
throw new ObTableRoutingWrongException();
21042096
}
2097+
} else if (result != null && result.isRoutingWrong()) {
2098+
if (result.isNeedRefreshMeta()) {
2099+
tableRoute.refreshMeta(tableName);
2100+
}
2101+
if (request instanceof ObTableAbstractOperationRequest) {
2102+
long tabletId = ((ObTableAbstractOperationRequest) request).getPartitionId();
2103+
tableRoute.refreshPartitionLocation(tableName, tabletId, null);
2104+
}
21052105
}
21062106
return result;
21072107
}
@@ -2365,7 +2365,7 @@ public ObPayload execute(final ObTableAbstractOperationRequest request) throws E
23652365
needRefreshTabletLocation = true;
23662366
if (ex instanceof ObTableNeedFetchMetaException) {
23672367
// Refresh table info
2368-
refreshMeta(request.getTableName());
2368+
tableRoute.refreshMeta(request.getTableName());
23692369
}
23702370
}
23712371
} else {

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

Lines changed: 33 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -119,42 +119,39 @@ public ObPayload invokeSync(final ObTableConnection conn, final ObPayload reques
119119
}
120120

121121
// decode ResultCode for response packet
122+
boolean isRoutingWrong = false;
123+
boolean isNeedRefreshMeta = false;
122124
ObRpcResultCode resultCode = new ObRpcResultCode();
123125
resultCode.decode(buf);
124-
// If response indicates the request is routed to wrong server, we should refresh the routing meta.
125-
if (!conn.getObTable().isEnableRerouting() && response.getHeader().isRoutingWrong()) {
126-
String errMessage = TraceUtil.formatTraceMessage(conn, request,
127-
"routed to the wrong server: [error code:" + resultCode.getRcode() + "]"
128-
+ resultCode.getErrMsg());
129-
logger.debug(errMessage);
130-
if (needFetchMeta(resultCode.getRcode(), resultCode.getPcode())) {
131-
throw new ObTableNeedFetchMetaException(errMessage, resultCode.getRcode());
132-
} else if (needFetchPartitionLocation(resultCode.getRcode())) {
133-
throw new ObTableRoutingWrongException(errMessage, resultCode.getRcode());
134-
} else {
135-
// Encountered an unexpected RoutingWrong error code,
136-
// possibly due to the client error code version being behind the observer's version.
137-
// Attempting a full refresh here
138-
// and delegating to the upper-level call to determine whether to throw the exception to the user based on the retry result.
139-
logger.warn("get unexpected error code: {}", errMessage);
140-
throw new ObTableNeedFetchMetaException(errMessage, resultCode.getRcode());
141-
}
142-
}
143-
if (resultCode.getRcode() != 0
144-
&& response.getHeader().getPcode() != Pcodes.OB_TABLE_API_MOVE) {
145-
String errMessage = TraceUtil.formatTraceMessage(conn, request,
146-
"meet exception: [error code:" + resultCode.getRcode() + "]"
147-
+ resultCode.getErrMsg());
148-
logger.debug(errMessage);
149-
if (needFetchMeta(resultCode.getRcode(), resultCode.getPcode())) {
150-
throw new ObTableNeedFetchMetaException(errMessage, resultCode.getRcode());
151-
} else if (needFetchPartitionLocation(resultCode.getRcode())) {
152-
throw new ObTableRoutingWrongException(errMessage, resultCode.getRcode());
153-
} else {
154-
logger.warn(errMessage);
155-
ExceptionUtil.throwObTableException(conn.getObTable().getIp(), conn
156-
.getObTable().getPort(), response.getHeader().getTraceId1(), response
157-
.getHeader().getTraceId0(), resultCode.getRcode(), resultCode.getErrMsg());
126+
logger.debug("[routing feedback] require_rerouting_: {}, need_refresh_kv_meta_: {}"
127+
, response.getHeader().isRoutingWrong(), response.getHeader().isNeedRefreshKvMeta());
128+
if (response.getHeader().getPcode() != Pcodes.OB_TABLE_API_MOVE) {
129+
if (resultCode.getRcode() != 0) {
130+
String errMessage = TraceUtil.formatTraceMessage(conn, request,
131+
"meet exception: [error code:" + resultCode.getRcode() + "]"
132+
+ resultCode.getErrMsg());
133+
logger.debug(errMessage);
134+
if (needFetchMeta(resultCode.getRcode(), resultCode.getPcode())) {
135+
throw new ObTableNeedFetchMetaException(errMessage, resultCode.getRcode());
136+
} else if (needFetchPartitionLocation(resultCode.getRcode())) {
137+
throw new ObTableRoutingWrongException(errMessage, resultCode.getRcode());
138+
} else {
139+
logger.warn(errMessage);
140+
ExceptionUtil.throwObTableException(conn.getObTable().getIp(), conn
141+
.getObTable().getPort(), response.getHeader().getTraceId1(), response
142+
.getHeader().getTraceId0(), resultCode.getRcode(), resultCode.getErrMsg());
143+
}
144+
} else if (resultCode.getRcode() == 0 && response.getHeader().isRoutingWrong()) {
145+
// if distributed capability is supported and enabled
146+
// just need to refresh table entry, no need to retry
147+
String errMessage = TraceUtil.formatTraceMessage(conn, request,
148+
"meet exception and retry successfully in server: [require_rerouting :" + response.getHeader().isRoutingWrong()
149+
+ ", need_refresh_kv_meta :" + response.getHeader().isNeedRefreshKvMeta() + "]");
150+
logger.debug(errMessage);
151+
isRoutingWrong = true;
152+
if (response.getHeader().isNeedRefreshKvMeta()) {
153+
isNeedRefreshMeta = true;
154+
}
158155
}
159156
}
160157

@@ -165,6 +162,8 @@ public ObPayload invokeSync(final ObTableConnection conn, final ObPayload reques
165162
.getHeader());
166163
payload.setSequence(response.getHeader().getTraceId1());
167164
payload.setUniqueId(response.getHeader().getTraceId0());
165+
payload.setIsRoutingWrong(isRoutingWrong);
166+
payload.setIsNeedRefreshMeta(isNeedRefreshMeta);
168167
} else {
169168
String errMessage = TraceUtil.formatTraceMessage(conn, response,
170169
"receive unexpected command code: " + response.getCmdCode().value());

‎src/main/java/com/alipay/oceanbase/rpc/protocol/packet/ObRpcPacketHeader.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -296,6 +296,14 @@ public void setRoutingWrong() {
296296
flag |= REQUIRE_REROUTING_FLAG;
297297
}
298298

299+
/*
300+
* need to refresh kv table meta
301+
* */
302+
public boolean isNeedRefreshKvMeta() {
303+
return (flag & IS_KV_REQUEST_FALG) != 0;
304+
}
305+
306+
299307
/*
300308
* Set stream next.
301309
*/

‎src/main/java/com/alipay/oceanbase/rpc/protocol/payload/AbstractPayload.java‎

Lines changed: 34 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,8 @@ public abstract class AbstractPayload implements ObPayload {
4242
private long uniqueId;
4343
private long sequence;
4444
private Integer channelId = null;
45+
private boolean isRoutingWrong = false; // flag means tableEntry location or meta need to be refreshed
46+
private boolean isNeedRefreshMeta = false; // flag means tableEntry meta need to be refreshed
4547
protected long tenantId = 1;
4648
private long version = 1;
4749
protected long timeout = RPC_OPERATION_TIMEOUT.getDefaultLong();
@@ -66,6 +68,38 @@ public long getTimeout() {
6668
return timeout;
6769
}
6870

71+
/*
72+
* Get isRoutingWrong
73+
* */
74+
@Override
75+
public boolean isRoutingWrong() {
76+
return this.isRoutingWrong;
77+
}
78+
79+
/*
80+
* Set isRoutingWrong
81+
* */
82+
@Override
83+
public void setIsRoutingWrong(boolean isRoutingWrong) {
84+
this.isRoutingWrong = isRoutingWrong;
85+
}
86+
87+
/*
88+
* Get isNeedRefreshMeta
89+
* */
90+
@Override
91+
public boolean isNeedRefreshMeta() {
92+
return this.isNeedRefreshMeta;
93+
}
94+
95+
/*
96+
* Set isNeedRefreshMeta
97+
* */
98+
@Override
99+
public void setIsNeedRefreshMeta(boolean isNeedRefreshMeta) {
100+
this.isNeedRefreshMeta = isNeedRefreshMeta;
101+
}
102+
69103
/*
70104
* Get version.
71105
*/

‎src/main/java/com/alipay/oceanbase/rpc/protocol/payload/ObPayload.java‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,14 @@ public interface ObPayload extends ObUnisVersion {
4141
*/
4242
long getTimeout();
4343

44+
boolean isRoutingWrong();
45+
46+
void setIsRoutingWrong(boolean isRoutingWrong);
47+
48+
boolean isNeedRefreshMeta();
49+
50+
void setIsNeedRefreshMeta(boolean isNeedRefreshMeta);
51+
4452
/*
4553
* set sequence
4654
*/

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

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -174,6 +174,12 @@ protected ObPayload commonExecute(ObTableClient client, Logger logger,
174174
.getServer().ipToString());
175175
throw new ObTableRoutingWrongException();
176176
}
177+
} else if (result != null && result.isRoutingWrong()) {
178+
TableEntry tableEntry = result.isNeedRefreshMeta() ?
179+
client.getOrRefreshTableEntry(indexTableName, true) :
180+
client.getOrRefreshTableEntry(indexTableName, false);
181+
long tabletId = client.getTabletIdByPartId(tableEntry, partIdWithIndex.getLeft());
182+
client.refreshTableLocationByTabletId(indexTableName, tabletId);
177183
}
178184
}
179185
client.resetExecuteContinuousFailureCount(indexTableName);

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

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -378,6 +378,12 @@ public void partitionExecute(ObTableOperationResult[] results,
378378
.ipToString());
379379
throw new ObTableRoutingWrongException();
380380
}
381+
} else if (result != null && result.isRoutingWrong()) {
382+
TableEntry entry = result.isNeedRefreshMeta() ?
383+
obTableClient.getOrRefreshTableEntry(tableName, true) :
384+
obTableClient.getOrRefreshTableEntry(tableName, false);
385+
long tabletId = obTableClient.getTabletIdByPartId(entry, originPartId);
386+
obTableClient.refreshTableLocationByTabletId(tableName, tabletId);
381387
}
382388
subObTableBatchOperationResult = (ObTableBatchOperationResult) result;
383389
obTableClient.resetExecuteContinuousFailureCount(tableName);

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

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -645,6 +645,13 @@ public void partitionExecute(ObTableSingleOpResult[] results,
645645
"Rerouting return IP is {}", moveResponse.getReplica().getServer().ipToString(), move.getReplica().getServer().ipToString());
646646
throw new ObTableRoutingWrongException();
647647
}
648+
} else if (result != null && result.isRoutingWrong()) {
649+
// retry successfully in server and need to refresh client cache
650+
if (result.isNeedRefreshMeta()) {
651+
obTableClient.getOrRefreshTableEntry(realTableName, true);
652+
}
653+
// TODO: 如果是不需要全部刷新地址的错误，全部刷新地址会降低效率。如何确定出错的 tablet_id 并刷新？
654+
obTableClient.refreshTabletLocationBatch(realTableName);
648655
}
649656
subLSOpResult = (ObTableLSOpResult) result;
650657
obTableClient.resetExecuteContinuousFailureCount(realTableName);

0 commit comments

Comments
 (0)