czy006 commented on code in PR #4261:
URL: https://github.com/apache/amoro/pull/4261#discussion_r3510233775
##########
amoro-ams/src/test/java/org/apache/amoro/server/TestDefaultOptimizingService.java:
##########
@@ -314,6 +317,31 @@ public void testAckAndCompleteTask() {
assertTaskCompleted(taskRuntime);
}
+ // Reproduces the EXACT path of issue #4235 end-to-end with the real
OptimizerKeeper: a live
+ // optimizer (the Toucher keeps heartbeating) polls a task but its ack is
delayed past
+ // OPTIMIZER_TASK_ACK_TIMEOUT (30s). The keeper, via the SCHEDULED +
ackTimeout branch of
+ // buildSuspendingPredication, resets the still-owned task to PLANNED. The
late ack then arrives
+ // and is rejected -- this is the "Task has been reset or not yet scheduled"
from the issue log,
+ // produced without any artificial retryTask() call.
+ @Test
+ public void testAckTimeoutResetThenLateAckRejected() throws
InterruptedException {
+ OptimizingTask task = optimizingService().pollTask(token, THREAD_ID);
+ Assertions.assertNotNull(task);
+ assertTaskStatus(TaskRuntime.Status.SCHEDULED); // polled but NOT acked
+
+ // ack-timeout is 30s; the optimizer stays alive (Toucher touches every
300ms), so this hits the
+ // SCHEDULED + ackTimeout branch rather than the optimizer-expired branch.
+ Thread.sleep(35000);
Review Comment:
on the default 30-second ack timeout and then sleeps for 35 seconds. This
adds a large fixed delay to the test suite and can still be brittle under slow
or noisy CI environments.
Suggested fix: reduce optimizer.task-ack-timeout for this test to a small
test-only value, such as `1-2 seconds`, and wait for the task status with
polling/await logic instead of a fixed 35-second sleep.
##########
amoro-ams/src/main/java/org/apache/amoro/server/optimizing/TaskRuntime.java:
##########
@@ -82,9 +86,22 @@ public SimpleFuture getCompletedFuture() {
}
void complete(OptimizerThread thread, OptimizingTaskResult result) {
+ // A completion for an already reset/rescheduled task is stale: the
execution it reports belongs
+ // to a round that the OptimizerKeeper has already torn down. Absorb it
gracefully instead of
+ // failing the (now illegal) state transition; the task will be
re-executed in its current
+ // round.
+ if (isStaleResponse(thread) || status != Status.ACKED) {
Review Comment:
`TaskRuntime.complete only` ignores a completion when
`isStaleResponse(thread) || status != ACKED`. If the old round’s completeTask
RPC is delayed until the same task has already gone through SCHEDULED -> ACKED
again on the same optimizer thread, then token, threadId, and status all match.
The old completion will pass the current check and be applied as the result of
the new round.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
