Skip to content

Commit 7721dc0

Browse files
author
张文领
committed
Fix the issue of optimizing tasks get stuck when stale ack arrives after task reset by OptimizerKeeper
1 parent d8f3793 commit 7721dc0

1 file changed

Lines changed: 20 additions & 6 deletions

File tree

amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/OptimizerExecutor.java

Lines changed: 20 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -234,16 +234,30 @@ private boolean ackTask(String amsUrl, OptimizingTask task) {
234234
amsUrl);
235235
return true;
236236
} catch (TException exception) {
237-
LOG.error(
238-
"Optimizer executor[{}] acknowledged task[{}] to AMS {} failed",
239-
threadId,
240-
task.getTaskId(),
241-
amsUrl,
242-
exception);
237+
if (isTaskResetOrNotScheduled(exception)) {
238+
LOG.warn(
239+
"Optimizer executor[{}] acknowledged task[{}] to AMS {} failed because "
240+
+ "task has been reset by OptimizerKeeper, this is expected and will be skipped",
241+
threadId,
242+
task.getTaskId(),
243+
amsUrl);
244+
} else {
245+
LOG.error(
246+
"Optimizer executor[{}] acknowledged task[{}] to AMS {} failed",
247+
threadId,
248+
task.getTaskId(),
249+
amsUrl,
250+
exception);
251+
}
243252
return false;
244253
}
245254
}
246255

256+
/** Check if the exception is caused by task reset by OptimizerKeeper. */
257+
private static boolean isTaskResetOrNotScheduled(TException exception) {
258+
return exception.getMessage() != null && exception.getMessage().contains("Task has been reset");
259+
}
260+
247261
protected OptimizingTaskResult executeTask(OptimizingTask task) {
248262
return executeTask(getConfig(), getThreadId(), task, LOG);
249263
}

0 commit comments

Comments
 (0)