Skip to content

Commit 0a3e212

Browse files
authored
Fix region-group cleanup: recover-safe submission + retry that actually re-runs the delete (#18097)
1 parent 660b0ad commit 0a3e212

6 files changed

Lines changed: 218 additions & 49 deletions

File tree

‎iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -623,7 +623,7 @@ public final class ProcedureMessages {
623623
public static final String PID_REMOVEREGIONGROUP_STATE_FAILED =
624624
"[pid{}][RemoveRegionGroup] state {} failed";
625625
public static final String PID_REMOVEREGIONGROUP_DELETE_REPLICA_FAILED =
626-
"[pid{}][RemoveRegionGroup] failed to delete a replica of region {} (attempt {}/{}), will retry. reason: {}";
626+
"[pid{}][RemoveRegionGroup] failed to delete a replica of region {} (attempt {}), will keep retrying until it is deleted. reason: {}";
627627
public static final String PID_REMOVEREGIONGROUP_SUCCESS_PROCEDURE_TOOK =
628628
"[pid{}][RemoveRegionGroup] success, region group {} has been deleted. Procedure took {} (started at {}).";
629629
public static final String PID_MIGRATEREGION_STARTED_WILL_BE_MIGRATED_FROM_DATANODE_TO =

‎iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -621,7 +621,7 @@ public final class ProcedureMessages {
621621
public static final String PID_REMOVEREGIONGROUP_STATE_FAILED =
622622
"[pid{}][RemoveRegionGroup] 状态 {} 失败";
623623
public static final String PID_REMOVEREGIONGROUP_DELETE_REPLICA_FAILED =
624-
"[pid{}][RemoveRegionGroup] 删除 region {} 的一个副本失败(第 {}/{} 次尝试),将重试。原因:{}";
624+
"[pid{}][RemoveRegionGroup] 删除 region {} 的一个副本失败(第 {} 次尝试),将持续重试直到删除成功。原因:{}";
625625
public static final String PID_REMOVEREGIONGROUP_SUCCESS_PROCEDURE_TOOK =
626626
"[pid{}][RemoveRegionGroup] 成功,region group {} 已删除。过程耗时 {}(开始于 {})。";
627627
public static final String PID_MIGRATEREGION_STARTED_WILL_BE_MIGRATED_FROM_DATANODE_TO =

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/CreateRegionGroupsProcedure.java‎

Lines changed: 25 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -108,8 +108,11 @@ protected Flow executeFromState(
108108
case SHUNT_REGION_REPLICAS:
109109
persistPlan = new CreateRegionGroupsPlan();
110110
final OfferRegionMaintainTasksPlan offerPlan = new OfferRegionMaintainTasksPlan();
111-
// RegionGroups that failed to reach a serving quorum are removed via a child
112-
// RemoveRegionGroupProcedure, which deletes every replica that did get created.
111+
// RegionGroups that failed to reach a serving quorum have their redundant (already-created)
112+
// replicas removed via an independent root RemoveRegionGroupProcedure. Submitting them as
113+
// root procedures (instead of children) keeps this procedure from waiting for or being
114+
// failed by the cleanup: each one retries forever until those replicas are deleted, while
115+
// this procedure proceeds to activate the region groups that did form a quorum.
113116
final List<RemoveRegionGroupProcedure> removeRegionGroupProcedures = new ArrayList<>();
114117
// Filter those RegionGroups that created successfully
115118
createRegionGroupsPlan
@@ -197,7 +200,26 @@ protected Flow executeFromState(
197200
LOGGER.warn(
198201
ConfigNodeMessages.FAILED_IN_THE_WRITE_API_EXECUTING_THE_CONSENSUS_LAYER_DUE, e);
199202
}
200-
removeRegionGroupProcedures.forEach(this::addChildProcedure);
203+
// Submit the redundant-replica cleanups as independent root procedures. This is
204+
// intentionally NOT guarded by isStateDeserialized(): the executor persists a procedure at
205+
// a state BEFORE that state's body has run (it advances the state on the previous cycle,
206+
// then may stop at the inter-state boundary on a leader switch — see
207+
// ProcedureExecutor#executeProcedure), so a recovery that lands on SHUNT_REGION_REPLICAS
208+
// means the submissions have NOT happened yet. Skipping them would leave the
209+
// already-created
210+
// replicas of sub-quorum region groups on disk with no cleanup and no partition-table
211+
// record
212+
// (the else branch above never persisted them). Re-submitting on recovery is safe instead:
213+
// the cleanups are recomputed from the serialized failedRegionReplicaSets, each gets a
214+
// fresh
215+
// procId and performs an idempotent delete, so a duplicate is harmless whereas a skip
216+
// leaks.
217+
removeRegionGroupProcedures.forEach(
218+
removeRegionGroupProcedure ->
219+
env.getConfigManager()
220+
.getProcedureManager()
221+
.getExecutor()
222+
.submitProcedure(removeRegionGroupProcedure));
201223
setNextState(CreateRegionGroupsState.REBALANCE_DATA_PARTITION_POLICY);
202224
break;
203225
case REBALANCE_DATA_PARTITION_POLICY:

‎iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/RemoveRegionGroupProcedure.java‎

Lines changed: 141 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -59,11 +59,21 @@
5959
* already-absent peer, so it works for a group of any size — including a sub-quorum group that
6060
* never finished forming. The DataNode runs the deletion asynchronously and this procedure polls
6161
* for the result, so a slow deletion is never wrongly reported as finished.
62+
*
63+
* <p>This procedure is submitted as an independent root procedure (not a child) by its callers,
64+
* which only enqueue the deletion and return immediately. It therefore owns the deletion end to
65+
* end: on any failure it retries the current replica forever (backing off between attempts) instead
66+
* of giving up, because there is no parent left to fall back to and the region's peer/data must not
67+
* be left on disk. Each genuine re-attempt uses a FRESH DataNode-side taskId (the DataNode dedups
68+
* by taskId and caches a terminal result forever, so reusing one taskId would make every retry a
69+
* no-op that never re-runs the delete); the in-flight taskId is persisted so a leader change
70+
* re-polls the same task rather than double-submitting. It carries its own {@link
71+
* TRegionReplicaSet} copy, so it can finish even after the caller has dropped the partition table,
72+
* and it survives ConfigNode leader change / restart.
6273
*/
6374
public class RemoveRegionGroupProcedure extends RegionOperationProcedure<RemoveRegionGroupState> {
6475
private static final Logger LOGGER = LoggerFactory.getLogger(RemoveRegionGroupProcedure.class);
6576

66-
private static final int MAX_DELETE_REPLICA_RETRY = 3;
6777
private static final long DELETE_REPLICA_RETRY_INTERVAL_MS = 5_000;
6878

6979
private TRegionReplicaSet regionReplicaSet;
@@ -75,10 +85,32 @@ public class RemoveRegionGroupProcedure extends RegionOperationProcedure<RemoveR
7585
// not finished deleting.
7686
private int currentReplicaIndex;
7787

78-
// Number of failed attempts on the replica at currentReplicaIndex. Transient: a leader change
79-
// restarts the retry budget for the current replica.
88+
// Number of failed attempts on the replica at currentReplicaIndex, used only for logging. Retries
89+
// are unbounded, so this is not a budget. Transient: a leader change restarts the counter for the
90+
// current replica.
8091
private transient int attemptedForCurrentReplica;
8192

93+
// Monotonic count of delete tasks this procedure has submitted, across all replicas. Persisted
94+
// and
95+
// only ever incremented. It is the low half of the DataNode-side taskId (see deleteTaskId): a
96+
// fresh value per genuine re-attempt makes the DataNode re-run the delete instead of replaying a
97+
// cached terminal result for a reused taskId (the DataNode dedups by taskId and never clears the
98+
// cache), which is the bug this fixes. It never resets, so every taskId this procedure emits is
99+
// distinct even across replicas and retries.
100+
private long deleteTaskSeq;
101+
102+
// Whether a delete task for the replica at currentReplicaIndex has already been submitted (and
103+
// thus
104+
// deleteTaskSeq already identifies an in-flight task to re-poll) rather than needing a fresh one.
105+
// Persisted so a leader change mid-attempt re-polls the SAME in-flight task instead of submitting
106+
// a
107+
// duplicate; cleared on success or when a terminal failure forces a fresh re-attempt.
108+
private boolean deleteTaskSubmitted;
109+
110+
// Bit budget for deleteTaskId(): sign bit (=> negative) + PROC_ID_BITS + SEQ_BITS must be <= 64.
111+
private static final int SEQ_BITS = 20;
112+
private static final int PROC_ID_BITS = 43;
113+
82114
public RemoveRegionGroupProcedure() {
83115
super();
84116
}
@@ -93,15 +125,27 @@ void setCurrentReplicaIndex(int currentReplicaIndex) {
93125
this.currentReplicaIndex = currentReplicaIndex;
94126
}
95127

128+
@TestOnly
129+
void setDeleteTaskState(long deleteTaskSeq, boolean deleteTaskSubmitted) {
130+
this.deleteTaskSeq = deleteTaskSeq;
131+
this.deleteTaskSubmitted = deleteTaskSubmitted;
132+
}
133+
134+
@TestOnly
135+
long deleteTaskIdForTest() {
136+
return deleteTaskId();
137+
}
138+
96139
@Override
97140
protected Flow executeFromState(ConfigNodeProcedureEnv env, RemoveRegionGroupState state)
98141
throws InterruptedException {
99142
final List<TDataNodeLocation> dataNodeLocations =
100143
regionReplicaSet == null ? null : regionReplicaSet.getDataNodeLocations();
101144
if (dataNodeLocations == null) {
102-
// A null replica set means deserialization failed; fail loudly instead of silently reporting
103-
// the group as deleted, otherwise the parent would drop the partition table while the region
104-
// data is still on disk.
145+
// A null replica set means deserialization failed. Retrying cannot recover the lost
146+
// locations,
147+
// so fail loudly instead of silently reporting the group as deleted (which would leave the
148+
// region's peer/data on disk with no record of where it lives).
105149
setFailure(
106150
new ProcedureException(ProcedureMessages.UNSUPPORTED_STATE + "missing regionReplicaSet"));
107151
return Flow.NO_MORE_STATE;
@@ -137,50 +181,62 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, RemoveRegionGroupSta
137181
regionId,
138182
simplifiedLocation(targetDataNode));
139183

140-
// deleteLocalPeer is idempotent (it tolerates an already-absent peer) and the DataNode
141-
// dedups by taskId, so re-submitting after a leader change or a retry is safe.
184+
// Start a fresh attempt (fresh taskId) unless we are resuming an already-submitted one
185+
// after
186+
// a leader change, in which case we re-poll the SAME task rather than submitting a
187+
// duplicate.
188+
if (!deleteTaskSubmitted) {
189+
deleteTaskSeq++;
190+
deleteTaskSubmitted = true;
191+
}
192+
final long deleteTaskId = deleteTaskId();
193+
194+
// deleteLocalPeer is idempotent (it tolerates an already-absent peer), and re-submitting
195+
// the
196+
// same taskId re-polls the same DataNode task, so resuming after a leader change is safe.
142197
final TSStatus submitStatus;
143198
final TRegionMigrateResult result;
144199
try {
145200
submitStatus =
146-
handler.submitDeleteOldRegionPeerTask(getProcId(), targetDataNode, regionId);
201+
handler.submitDeleteOldRegionPeerTask(deleteTaskId, targetDataNode, regionId);
147202
setKillPoint(state);
148203
if (submitStatus.getCode() != SUCCESS_STATUS.getStatusCode()) {
149-
return retryCurrentReplicaOrFail(
204+
return retryCurrentReplica(
150205
String.format(
151206
"submit delete task for region %s to DataNode %s failed: %s",
152207
regionId, simplifiedLocation(targetDataNode), submitStatus));
153208
}
154-
result = handler.waitTaskFinish(getProcId(), targetDataNode);
209+
result = handler.waitTaskFinish(deleteTaskId, targetDataNode);
155210
} catch (InterruptedException e) {
156211
throw e;
157212
} catch (Exception e) {
158213
LOGGER.error(ProcedureMessages.PID_REMOVEREGIONGROUP_STATE_FAILED, getProcId(), state, e);
159-
return retryCurrentReplicaOrFail(
214+
return retryCurrentReplica(
160215
String.format(
161216
"delete region %s from DataNode %s threw %s",
162217
regionId, simplifiedLocation(targetDataNode), e));
163218
}
164219

165220
switch (result.getTaskStatus()) {
166221
case SUCCESS:
167-
// Advance to the next replica with a fresh retry budget.
222+
// Advance to the next replica with a fresh retry counter and a fresh delete task.
168223
currentReplicaIndex++;
169224
attemptedForCurrentReplica = 0;
225+
deleteTaskSubmitted = false;
170226
setNextState(RemoveRegionGroupState.DELETE_REGION_REPLICAS);
171227
return Flow.HAS_MORE_STATE;
172228
case PROCESSING:
173229
// waitTaskFinish() only returns PROCESSING when its polling loop was interrupted, i.e.
174230
// this ConfigNode is shutting down / losing leadership. The delete task is still
175-
// running
176-
// on the DataNode, so persist and re-poll after recovery: stay on this replica without
177-
// advancing it and without consuming a retry attempt.
231+
// running on the DataNode, so persist and re-poll after recovery: stay on this replica
232+
// without advancing it, without consuming a retry attempt, and keeping deleteTaskSeq /
233+
// deleteTaskSubmitted so the re-poll targets the same in-flight task.
178234
setNextState(RemoveRegionGroupState.DELETE_REGION_REPLICAS);
179235
return Flow.HAS_MORE_STATE;
180236
case TASK_NOT_EXIST:
181237
case FAIL:
182238
default:
183-
return retryCurrentReplicaOrFail(
239+
return retryCurrentReplica(
184240
String.format(
185241
"delete region %s from DataNode %s, task status is %s",
186242
regionId, simplifiedLocation(targetDataNode), result.getTaskStatus()));
@@ -192,30 +248,57 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, RemoveRegionGroupSta
192248
}
193249

194250
/**
195-
* Retry the replica at {@link #currentReplicaIndex} after a backoff, or fail the whole procedure
196-
* once the per-replica retry budget is exhausted. Failing (rather than skipping the replica)
197-
* keeps the parent from dropping the partition table while a region's peer/data is still on disk.
251+
* Retry the replica at {@link #currentReplicaIndex} after a backoff. This procedure never gives
252+
* up on a replica: because it is submitted as an independent root procedure, there is no parent
253+
* to fall back to, and skipping or failing would leave the region's peer/data on disk. So it
254+
* backs off and re-runs the same state until the replica is deleted, which eventually succeeds
255+
* once the target DataNode is reachable: the delete is idempotent, and clearing {@link
256+
* #deleteTaskSubmitted} here makes the next attempt use a FRESH DataNode-side taskId (a new
257+
* {@link #deleteTaskSeq}), so the DataNode actually re-executes the delete instead of returning a
258+
* cached terminal result for the previous taskId.
198259
*/
199-
private Flow retryCurrentReplicaOrFail(String reason) throws InterruptedException {
260+
private Flow retryCurrentReplica(String reason) throws InterruptedException {
200261
attemptedForCurrentReplica++;
201-
if (attemptedForCurrentReplica <= MAX_DELETE_REPLICA_RETRY) {
202-
LOGGER.warn(
203-
ProcedureMessages.PID_REMOVEREGIONGROUP_DELETE_REPLICA_FAILED,
204-
getProcId(),
205-
regionId,
206-
attemptedForCurrentReplica,
207-
MAX_DELETE_REPLICA_RETRY + 1,
208-
reason);
209-
Thread.sleep(DELETE_REPLICA_RETRY_INTERVAL_MS);
210-
setNextState(RemoveRegionGroupState.DELETE_REGION_REPLICAS);
211-
return Flow.HAS_MORE_STATE;
262+
LOGGER.warn(
263+
ProcedureMessages.PID_REMOVEREGIONGROUP_DELETE_REPLICA_FAILED,
264+
getProcId(),
265+
regionId,
266+
attemptedForCurrentReplica,
267+
reason);
268+
// Force a fresh delete task on the next attempt so the DataNode re-runs the delete rather than
269+
// replaying a cached FAIL/SUCCESS for this taskId.
270+
deleteTaskSubmitted = false;
271+
Thread.sleep(DELETE_REPLICA_RETRY_INTERVAL_MS);
272+
setNextState(RemoveRegionGroupState.DELETE_REGION_REPLICAS);
273+
return Flow.HAS_MORE_STATE;
274+
}
275+
276+
/**
277+
* The DataNode-side taskId for the current attempt, derived from this procedure's (globally
278+
* unique, consensus-replicated) procId and its monotonic {@link #deleteTaskSeq}. It is packed
279+
* into the NEGATIVE i64 space, which is disjoint from every real procId (all {@code >= 0}); other
280+
* region-maintain procedures (add/remove peer) use {@code getProcId()} directly as the taskId
281+
* against the same DataNode task map, so a negative id can never collide with theirs. Unlike
282+
* minting from the procedure-store id allocator, this needs nothing extra replicated: procId is
283+
* already replicated and deleteTaskSeq is persisted with this procedure, so the taskId is stable
284+
* across a leader change and never regresses.
285+
*
286+
* <p>Layout: sign bit set (=> negative) | {@value PROC_ID_BITS} bits of procId | {@value
287+
* SEQ_BITS} bits of deleteTaskSeq. The bounds are astronomically beyond any real cluster (a
288+
* procId needs 2^43 procedures; a single group delete needs 2^20 retries), and are asserted
289+
* rather than silently wrapped so a violation fails the procedure loudly instead of emitting a
290+
* colliding id.
291+
*/
292+
private long deleteTaskId() {
293+
final long procId = getProcId();
294+
if (procId < 0 || procId >= (1L << PROC_ID_BITS) || deleteTaskSeq >= (1L << SEQ_BITS)) {
295+
throw new IllegalStateException(
296+
String.format(
297+
"cannot derive a collision-free delete taskId: procId=%d, deleteTaskSeq=%d exceed the "
298+
+ "%d/%d-bit budget",
299+
procId, deleteTaskSeq, PROC_ID_BITS, SEQ_BITS));
212300
}
213-
setFailure(
214-
new ProcedureException(
215-
String.format(
216-
"[pid%d][RemoveRegionGroup] gave up after %d attempts: %s",
217-
getProcId(), attemptedForCurrentReplica, reason)));
218-
return Flow.NO_MORE_STATE;
301+
return Long.MIN_VALUE | (procId << SEQ_BITS) | deleteTaskSeq;
219302
}
220303

221304
@Override
@@ -243,6 +326,11 @@ public void serialize(DataOutputStream stream) throws IOException {
243326
super.serialize(stream);
244327
ThriftCommonsSerDeUtils.serializeTRegionReplicaSet(regionReplicaSet, stream);
245328
ReadWriteIOUtils.write(currentReplicaIndex, stream);
329+
// Persist the delete-task cursor so a leader change re-derives the SAME in-flight taskId and
330+
// re-polls it (deleteTaskSubmitted == true) instead of submitting a duplicate, and so the
331+
// monotonic deleteTaskSeq never regresses.
332+
ReadWriteIOUtils.write(deleteTaskSeq, stream);
333+
ReadWriteIOUtils.write(deleteTaskSubmitted, stream);
246334
}
247335

248336
@Override
@@ -252,6 +340,14 @@ public void deserialize(ByteBuffer byteBuffer) {
252340
regionReplicaSet = ThriftCommonsSerDeUtils.deserializeTRegionReplicaSet(byteBuffer);
253341
regionId = regionReplicaSet.getRegionId();
254342
currentReplicaIndex = ReadWriteIOUtils.readInt(byteBuffer);
343+
// deleteTaskSeq/deleteTaskSubmitted were appended after the first version of this procedure.
344+
// That first version only ever existed on the unreleased branch that added this procedure
345+
// (never in a release), but a dev/CI cluster could persist a blob without these trailing
346+
// fields; tolerate it by defaulting to "no in-flight task" instead of reading past the end.
347+
if (byteBuffer.hasRemaining()) {
348+
deleteTaskSeq = ReadWriteIOUtils.readLong(byteBuffer);
349+
deleteTaskSubmitted = ReadWriteIOUtils.readBool(byteBuffer);
350+
}
255351
} catch (ThriftSerDeException e) {
256352
LOGGER.error(ProcedureMessages.ERROR_IN_DESERIALIZE, this.getClass(), e);
257353
}
@@ -264,12 +360,14 @@ public boolean equals(Object obj) {
264360
}
265361
RemoveRegionGroupProcedure procedure = (RemoveRegionGroupProcedure) obj;
266362
return this.currentReplicaIndex == procedure.currentReplicaIndex
363+
&& this.deleteTaskSeq == procedure.deleteTaskSeq
364+
&& this.deleteTaskSubmitted == procedure.deleteTaskSubmitted
267365
&& Objects.equals(this.regionReplicaSet, procedure.regionReplicaSet);
268366
}
269367

270368
@Override
271369
public int hashCode() {
272-
return Objects.hash(regionReplicaSet, currentReplicaIndex);
370+
return Objects.hash(regionReplicaSet, currentReplicaIndex, deleteTaskSeq, deleteTaskSubmitted);
273371
}
274372

275373
@Override
@@ -279,6 +377,10 @@ public String toString() {
279377
+ regionReplicaSet
280378
+ ", currentReplicaIndex="
281379
+ currentReplicaIndex
380+
+ ", deleteTaskSeq="
381+
+ deleteTaskSeq
382+
+ ", deleteTaskSubmitted="
383+
+ deleteTaskSubmitted
282384
+ '}';
283385
}
284386
}

0 commit comments

Comments
 (0)