Skip to content

Commit 3fbbe5f

Browse files
CRZbulabulaCRZbulabula
authored
[To dev/1.3] Harden RegionMaintainer retries and region operation atomicity (#18480)
1 parent fda8031 commit 3fbbe5f

16 files changed

Lines changed: 855 additions & 298 deletions

File tree

‎iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -166,6 +166,8 @@ public enum TSStatusCode {
166166
RECONSTRUCT_REGION_ERROR(908),
167167
EXTEND_REGION_ERROR(909),
168168
REMOVE_REGION_PEER_ERROR(910),
169+
REGION_ALREADY_EXISTS(911),
170+
REGION_NOT_EXIST(912),
169171

170172
// Cluster Manager
171173
ADD_CONFIGNODE_ERROR(1000),

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeTSStatusRPCHandler.java‎

Lines changed: 13 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,7 @@ public void onComplete(TSStatus response) {
5252
// Put response
5353
responseMap.put(requestId, response);
5454

55-
if (response.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
55+
if (isRequestCompleted(requestType, response)) {
5656
// Remove only if success
5757
nodeLocationMap.remove(requestId);
5858
LOGGER.info("Successfully {} on DataNode: {}", requestType, formattedTargetLocation);
@@ -68,6 +68,18 @@ public void onComplete(TSStatus response) {
6868
countDownLatch.countDown();
6969
}
7070

71+
static boolean isRequestCompleted(CnToDnAsyncRequestType requestType, TSStatus response) {
72+
if (response.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
73+
return true;
74+
}
75+
if (requestType == CnToDnAsyncRequestType.DELETE_REGION) {
76+
return response.getCode() == TSStatusCode.REGION_NOT_EXIST.getStatusCode();
77+
}
78+
return (requestType == CnToDnAsyncRequestType.CREATE_DATA_REGION
79+
|| requestType == CnToDnAsyncRequestType.CREATE_SCHEMA_REGION)
80+
&& response.getCode() == TSStatusCode.REGION_ALREADY_EXISTS.getStatusCode();
81+
}
82+
7183
@Override
7284
public void onError(Exception e) {
7385
String errorMsg =

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/partition/PartitionManager.java‎

Lines changed: 182 additions & 184 deletions
Large diffs are not rendered by default.

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteDatabaseProcedure.java‎

Lines changed: 74 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -112,38 +112,17 @@ protected Flow executeFromState(
112112
deleteDatabaseSchema.getName());
113113

114114
// Submit RegionDeleteTasks
115-
OfferRegionMaintainTasksPlan dataRegionDeleteTaskOfferPlan =
116-
new OfferRegionMaintainTasksPlan();
117115
List<TRegionReplicaSet> regionReplicaSets =
118116
env.getAllReplicaSets(deleteDatabaseSchema.getName());
119-
List<TRegionReplicaSet> schemaRegionReplicaSets = new ArrayList<>();
117+
OfferRegionMaintainTasksPlan regionDeleteTaskOfferPlan =
118+
buildDataRegionDeleteTaskOfferPlan(regionReplicaSets);
119+
List<TRegionReplicaSet> schemaRegionReplicaSets =
120+
getSchemaRegionReplicaSets(regionReplicaSets);
120121
regionReplicaSets.forEach(
121-
regionReplicaSet -> {
122-
// Clear heartbeat cache along the way
123-
env.getConfigManager()
124-
.getLoadManager()
125-
.removeRegionGroupRelatedCache(regionReplicaSet.getRegionId());
126-
127-
if (regionReplicaSet
128-
.getRegionId()
129-
.getType()
130-
.equals(TConsensusGroupType.SchemaRegion)) {
131-
schemaRegionReplicaSets.add(regionReplicaSet);
132-
} else {
133-
regionReplicaSet
134-
.getDataNodeLocations()
135-
.forEach(
136-
targetDataNode ->
137-
dataRegionDeleteTaskOfferPlan.appendRegionMaintainTask(
138-
new RegionDeleteTask(
139-
targetDataNode, regionReplicaSet.getRegionId())));
140-
}
141-
});
142-
143-
if (!dataRegionDeleteTaskOfferPlan.getRegionMaintainTaskList().isEmpty()) {
144-
// submit async data region delete task
145-
env.getConfigManager().getConsensusManager().write(dataRegionDeleteTaskOfferPlan);
146-
}
122+
regionReplicaSet ->
123+
env.getConfigManager()
124+
.getLoadManager()
125+
.removeRegionGroupRelatedCache(regionReplicaSet.getRegionId()));
147126

148127
// try sync delete schemaengine region
149128
DataNodeAsyncRequestContext<TConsensusGroupId, TSStatus> asyncClientHandler =
@@ -166,7 +145,7 @@ protected Flow executeFromState(
166145
.sendAsyncRequestWithRetry(asyncClientHandler);
167146
for (Map.Entry<Integer, TSStatus> entry :
168147
asyncClientHandler.getResponseMap().entrySet()) {
169-
if (entry.getValue().getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
148+
if (isRegionDeleteCompleted(entry.getValue())) {
170149
LOG.info(
171150
"[DeleteDatabaseProcedure] Successfully delete SchemaRegion[{}] on {}",
172151
asyncClientHandler.getRequest(entry.getKey()),
@@ -181,16 +160,17 @@ protected Flow executeFromState(
181160
}
182161

183162
if (!schemaRegionDeleteTaskMap.isEmpty()) {
184-
// submit async schemaengine region delete task for failed sync execution
185-
OfferRegionMaintainTasksPlan schemaRegionDeleteTaskOfferPlan =
186-
new OfferRegionMaintainTasksPlan();
187-
schemaRegionDeleteTaskMap
188-
.values()
189-
.forEach(schemaRegionDeleteTaskOfferPlan::appendRegionMaintainTask);
190-
env.getConfigManager().getConsensusManager().write(schemaRegionDeleteTaskOfferPlan);
163+
// submit async schemaengine region delete tasks for failed sync executions
164+
appendFailedSchemaRegionDeleteTasks(
165+
regionDeleteTaskOfferPlan, schemaRegionDeleteTaskMap);
191166
}
192167
}
193168

169+
if (!offerRegionDeleteTasks(env, regionDeleteTaskOfferPlan)) {
170+
setNextState(DeleteStorageGroupState.DELETE_DATABASE_SCHEMA);
171+
return Flow.HAS_MORE_STATE;
172+
}
173+
194174
env.getConfigManager()
195175
.getLoadManager()
196176
.clearDataPartitionPolicyTable(deleteDatabaseSchema.getName());
@@ -212,6 +192,8 @@ protected Flow executeFromState(
212192
} else if (getCycles() > RETRY_THRESHOLD) {
213193
setFailure(
214194
new ProcedureException("[DeleteDatabaseProcedure] Delete DatabaseSchema failed"));
195+
} else {
196+
setNextState(DeleteStorageGroupState.DELETE_DATABASE_SCHEMA);
215197
}
216198
}
217199
} catch (ConsensusException | TException | IOException e) {
@@ -230,12 +212,67 @@ protected Flow executeFromState(
230212
e);
231213
if (getCycles() > RETRY_THRESHOLD) {
232214
setFailure(new ProcedureException("[DeleteDatabaseProcedure] State stuck at " + state));
215+
} else {
216+
setNextState(state);
233217
}
234218
}
235219
}
236220
return Flow.HAS_MORE_STATE;
237221
}
238222

223+
static boolean isRegionDeleteCompleted(TSStatus status) {
224+
return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
225+
|| status.getCode() == TSStatusCode.REGION_NOT_EXIST.getStatusCode();
226+
}
227+
228+
static OfferRegionMaintainTasksPlan buildDataRegionDeleteTaskOfferPlan(
229+
List<TRegionReplicaSet> regionReplicaSets) {
230+
OfferRegionMaintainTasksPlan offerPlan = new OfferRegionMaintainTasksPlan();
231+
regionReplicaSets.stream()
232+
.filter(
233+
regionReplicaSet ->
234+
regionReplicaSet.getRegionId().getType() == TConsensusGroupType.DataRegion)
235+
.forEach(
236+
regionReplicaSet ->
237+
regionReplicaSet
238+
.getDataNodeLocations()
239+
.forEach(
240+
targetDataNode ->
241+
offerPlan.appendRegionMaintainTask(
242+
new RegionDeleteTask(
243+
targetDataNode, regionReplicaSet.getRegionId()))));
244+
return offerPlan;
245+
}
246+
247+
private static List<TRegionReplicaSet> getSchemaRegionReplicaSets(
248+
List<TRegionReplicaSet> regionReplicaSets) {
249+
List<TRegionReplicaSet> schemaRegionReplicaSets = new ArrayList<>();
250+
regionReplicaSets.stream()
251+
.filter(
252+
regionReplicaSet ->
253+
regionReplicaSet.getRegionId().getType() == TConsensusGroupType.SchemaRegion)
254+
.forEach(schemaRegionReplicaSets::add);
255+
return schemaRegionReplicaSets;
256+
}
257+
258+
static void appendFailedSchemaRegionDeleteTasks(
259+
OfferRegionMaintainTasksPlan offerPlan,
260+
Map<Integer, RegionDeleteTask> failedSchemaRegionDeleteTasks) {
261+
failedSchemaRegionDeleteTasks.values().forEach(offerPlan::appendRegionMaintainTask);
262+
}
263+
264+
private boolean offerRegionDeleteTasks(
265+
ConfigNodeProcedureEnv env, OfferRegionMaintainTasksPlan offerPlan)
266+
throws ConsensusException {
267+
return offerPlan.getRegionMaintainTaskList().isEmpty()
268+
|| isRegionDeleteTaskOfferSuccessful(
269+
env.getConfigManager().getConsensusManager().write(offerPlan));
270+
}
271+
272+
static boolean isRegionDeleteTaskOfferSuccessful(TSStatus status) {
273+
return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode();
274+
}
275+
239276
@Override
240277
protected void rollbackState(ConfigNodeProcedureEnv env, DeleteStorageGroupState state)
241278
throws IOException, InterruptedException {

‎iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeTSStatusRPCHandlerTest.java‎

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,63 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
package org.apache.iotdb.confignode.client.async.handlers.rpc;
21+
22+
import org.apache.iotdb.common.rpc.thrift.TSStatus;
23+
import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType;
24+
import org.apache.iotdb.rpc.TSStatusCode;
25+
26+
import org.junit.Test;
27+
28+
import static org.junit.Assert.assertFalse;
29+
import static org.junit.Assert.assertTrue;
30+
31+
public class DataNodeTSStatusRPCHandlerTest {
32+
33+
@Test
34+
public void testRegionOperationTerminalStatuses() {
35+
assertTrue(
36+
DataNodeTSStatusRPCHandler.isRequestCompleted(
37+
CnToDnAsyncRequestType.CREATE_DATA_REGION, status(TSStatusCode.SUCCESS_STATUS)));
38+
assertTrue(
39+
DataNodeTSStatusRPCHandler.isRequestCompleted(
40+
CnToDnAsyncRequestType.CREATE_DATA_REGION, status(TSStatusCode.REGION_ALREADY_EXISTS)));
41+
assertTrue(
42+
DataNodeTSStatusRPCHandler.isRequestCompleted(
43+
CnToDnAsyncRequestType.CREATE_SCHEMA_REGION,
44+
status(TSStatusCode.REGION_ALREADY_EXISTS)));
45+
assertTrue(
46+
DataNodeTSStatusRPCHandler.isRequestCompleted(
47+
CnToDnAsyncRequestType.DELETE_REGION, status(TSStatusCode.REGION_NOT_EXIST)));
48+
49+
assertFalse(
50+
DataNodeTSStatusRPCHandler.isRequestCompleted(
51+
CnToDnAsyncRequestType.CREATE_SCHEMA_REGION, status(TSStatusCode.CREATE_REGION_ERROR)));
52+
assertFalse(
53+
DataNodeTSStatusRPCHandler.isRequestCompleted(
54+
CnToDnAsyncRequestType.DELETE_REGION, status(TSStatusCode.DELETE_REGION_ERROR)));
55+
assertFalse(
56+
DataNodeTSStatusRPCHandler.isRequestCompleted(
57+
CnToDnAsyncRequestType.SET_TTL, status(TSStatusCode.REGION_ALREADY_EXISTS)));
58+
}
59+
60+
private static TSStatus status(TSStatusCode statusCode) {
61+
return new TSStatus(statusCode.getStatusCode());
62+
}
63+
}

0 commit comments

Comments
 (0)