You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@amoro.apache.org by cz...@apache.org on 2026/07/15 15:21:08 UTC
(amoro) branch master updated: [AMORO-4235] Recognize stale optimizer responses on the AMS side (#4261)
This is an automated email from the ASF dual-hosted git repository.
czy006 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new 27ae8d55f [AMORO-4235] Recognize stale optimizer responses on the AMS side (#4261)
27ae8d55f is described below
commit 27ae8d55f2d300acb0d123602d90928ab7349b6b
Author: Jiwon Park <jp...@outlook.kr>
AuthorDate: Thu Jul 16 00:21:04 2026 +0900
[AMORO-4235] Recognize stale optimizer responses on the AMS side (#4261)
* [AMORO-4235] Recognize stale optimizer responses on the AMS side
When the OptimizerKeeper (or resetStaleTasksForThread on re-poll) resets a
task that is still executing, the optimizer's in-flight ack/complete arrives
after the task's token has been cleared. TaskRuntime.validThread then threw
"Task has been reset or not yet scheduled", surfacing as an ERROR; for
completion it also broke the SCHEDULED -> SUCCESS transition
(IllegalTaskStateException).
The AMS now recognizes stale responses by (token, threadId, status):
- ack: rejected outside the transaction so the optimizer skips the obsolete
round, without the misleading "failed to commit transaction" error.
- complete: ignored gracefully, since the reported run belongs to a
torn-down round.
This fixes the rejection of valid optimizer responses. The permanent-stuck /
uncancelable symptom in the issue could not be reproduced from the excerpt
logs and may need the full log to confirm -- hence "Relates to" rather than
"Close".
Tests:
- TestOptimizingQueue: unit reproductions via the pollTask path (multi-task
stale ack, single-task stale completion).
- TestDefaultOptimizingService: end-to-end reproduction driving the real
OptimizerKeeper ack-timeout reset of a live optimizer's task.
Relates to #4235
Signed-off-by: Jiwon Park <jp...@outlook.kr>
* [AMORO-4235] Fix testAckAndCompleteTask for absorbed stale completion
Completing a task before ack now produces no exception (the AMS absorbs it
as a stale response, indistinguishable from a completion for a task reset
and re-scheduled to the same thread), so assert the task stays SCHEDULED
instead of expecting IllegalTaskStateException.
Signed-off-by: Jiwon Park <jp...@outlook.kr>
* [AMORO-4235] Apply spotless formatting
Signed-off-by: Jiwon Park <jp...@outlook.kr>
* [AMORO-4235] Lower test ack-timeout and poll for status instead of sleeping 35s
Address review feedback: testAckTimeoutResetThenLateAckRejected relied on the
default 30s optimizer.task-ack-timeout and a fixed Thread.sleep(35000).
- Override OPTIMIZER_TASK_ACK_TIMEOUT to 5s in AMSServiceTestBase. It must
stay above optimizer.polling-timeout (3s): a blocking pollTask waiting out
its full timeout would otherwise pick up the task the keeper reset by
ack-timeout in the meantime (this broke testPollTaskThreeTimes at 2s).
- Replace the fixed sleep with waitForTaskStatus polling every 100ms, so the
test finishes right after the keeper reset lands (~6s instead of 35s).
Signed-off-by: Jiwon Park <jp...@outlook.kr>
* [AMORO-4235] Stamp an attempt id on schedule and ignore completions of previous attempts
A completion delayed until the same task has been reset, re-scheduled to
the same optimizer thread and re-acked matches (token, threadId, status)
and was accepted as the result of the new attempt.
Every schedule now stamps a monotonically increasing attempt id into the
task properties (reaching the optimizer via the existing
OptimizingTask.properties map), the optimizer echoes it back in the
result summary, and complete() ignores a completion whose echoed attempt
id does not match the current one. Results from older optimizers carry
no attempt id and fall back to the previous checks.
Signed-off-by: Jiwon Park <jp...@outlook.kr>
---------
Signed-off-by: Jiwon Park <jp...@outlook.kr>
Co-authored-by: ConradJam <ja...@gmail.com>
---
.../amoro/server/optimizing/TaskRuntime.java | 98 ++++++++++++++++--
.../apache/amoro/server/AMSServiceTestBase.java | 3 +
.../amoro/server/TestDefaultOptimizingService.java | 47 ++++++++-
.../server/optimizing/TestOptimizingQueue.java | 115 +++++++++++++++++++++
.../apache/amoro/process/StagedTaskDescriptor.java | 4 +
.../apache/amoro/optimizing/TaskProperties.java | 7 ++
.../amoro/optimizer/common/OptimizerExecutor.java | 22 ++++
.../optimizer/common/TestOptimizerExecutor.java | 36 +++++++
8 files changed, 317 insertions(+), 15 deletions(-)
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/TaskRuntime.java b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/TaskRuntime.java
index d0c3f9f7c..fc7e67877 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/TaskRuntime.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/optimizing/TaskRuntime.java
@@ -24,6 +24,7 @@ import org.apache.amoro.api.OptimizingTaskId;
import org.apache.amoro.api.OptimizingTaskResult;
import org.apache.amoro.exception.IllegalTaskStateException;
import org.apache.amoro.exception.TaskRuntimeException;
+import org.apache.amoro.optimizing.TaskProperties;
import org.apache.amoro.process.SimpleFuture;
import org.apache.amoro.process.StagedTaskDescriptor;
import org.apache.amoro.server.AmoroServiceConstants;
@@ -33,12 +34,17 @@ import org.apache.amoro.server.resource.OptimizerThread;
import org.apache.amoro.shade.guava32.com.google.common.base.MoreObjects;
import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableMap;
import org.apache.amoro.shade.guava32.com.google.common.collect.ImmutableSet;
+import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import java.util.Map;
import java.util.Set;
public class TaskRuntime<T extends StagedTaskDescriptor<?, ?, ?>> extends StatedPersistentBase {
+ private static final Logger LOG = LoggerFactory.getLogger(TaskRuntime.class);
+
private final SimpleFuture future = new SimpleFuture();
private final TaskStatusMachine statusMachine = new TaskStatusMachine();
private OptimizingTaskId taskId;
@@ -82,9 +88,25 @@ public class TaskRuntime<T extends StagedTaskDescriptor<?, ?, ?>> extends Stated
}
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 || isStaleAttempt(result)) {
+ LOG.warn(
+ "Ignoring stale completion for task {} from optimizer thread {}, current status {}, "
+ + "owner {}, attempt {} (reported attempt {}). The task was likely reset by the "
+ + "OptimizerKeeper and will be re-executed.",
+ taskId,
+ thread,
+ status,
+ getResourceDesc(),
+ currentAttemptId(),
+ reportedAttemptId(result));
+ return;
+ }
invokeConsistency(
() -> {
- validThread(thread);
if (result.getErrorMessage() != null) {
statusMachine.accept(Status.FAILED);
failReason = result.getErrorMessage();
@@ -128,14 +150,27 @@ public class TaskRuntime<T extends StagedTaskDescriptor<?, ?, ?>> extends Stated
token = thread.getToken();
threadId = thread.getThreadId();
startTime = System.currentTimeMillis();
+ stampNextAttemptId();
persistTaskRuntime();
});
}
void ack(OptimizerThread thread) {
+ // If the task has been reset/rescheduled, reject the stale ack so the optimizer skips executing
+ // this obsolete round (OptimizerExecutor treats the failure as "skip"). Thrown outside the
+ // transaction to avoid a misleading "failed to commit transaction" error for this expected
+ // case.
+ if (isStaleResponse(thread) || status != Status.SCHEDULED) {
+ LOG.warn(
+ "Rejecting stale ack for task {} from optimizer thread {}: current status {}, owner {}.",
+ taskId,
+ thread,
+ status,
+ getResourceDesc());
+ throw new TaskRuntimeException("Task has been reset or not yet scheduled, taskId:%s", taskId);
+ }
invokeConsistency(
() -> {
- validThread(thread);
statusMachine.accept(Status.ACKED);
persistTaskRuntime();
});
@@ -247,15 +282,56 @@ public class TaskRuntime<T extends StagedTaskDescriptor<?, ?, ?>> extends Stated
.toString();
}
- private void validThread(OptimizerThread thread) {
- if (token == null) {
- throw new TaskRuntimeException("Task has been reset or not yet scheduled, taskId:%s", taskId);
- }
- if (!thread.getToken().equals(getToken()) || thread.getThreadId() != threadId) {
- throw new TaskRuntimeException(
- "The optimizer thread does not match, the thread in the task is OptimizerThread(token=%s, threadId=%s), and the thread in the request is OptimizerThread(token=%s, threadId=%s).",
- getToken(), threadId, thread.getToken(), thread.getThreadId());
- }
+ /**
+ * Detects a response from an optimizer round the task no longer belongs to: it was reset (token
+ * cleared) or has since been rescheduled to a different owner. Reads token/threadId outside
+ * {@link #invokeConsistency}, so callers must hold the owning TableOptimizingProcess lock, under
+ * which OptimizingQueue serializes ack/complete/reset/poll.
+ */
+ private boolean isStaleResponse(OptimizerThread thread) {
+ return token == null || !thread.getToken().equals(token) || thread.getThreadId() != threadId;
+ }
+
+ /**
+ * Detects a completion from a previous attempt of this task, which {@link #isStaleResponse}
+ * cannot: once a reset task is re-scheduled to the same optimizer thread and re-acked, (token,
+ * threadId, status) all match again. Optimizers echo the attempt id they received with the task
+ * back in the result summary; a result from an older optimizer carries none and is checked by
+ * (token, threadId, status) alone.
+ */
+ private boolean isStaleAttempt(OptimizingTaskResult result) {
+ String reported = reportedAttemptId(result);
+ return reported != null && !reported.equals(currentAttemptId());
+ }
+
+ /**
+ * Advances the attempt id stamped into the task properties, which reach the optimizer via {@link
+ * #extractProtocolTask()}. The properties map is replaced instead of mutated, because
+ * extractProtocolTask hands the live reference to thrift serialization outside the process lock.
+ * The id is persisted with the task, so it stays monotonic across AMS restarts; if the enclosing
+ * transaction rolls back, the in-memory increment survives (properties are not a {@link
+ * StateField}), which merely skips a value and never reuses one.
+ */
+ private void stampNextAttemptId() {
+ String current = currentAttemptId();
+ long next = current == null ? 1 : Long.parseLong(current) + 1;
+ Map<String, String> properties =
+ taskDescriptor.getProperties() == null
+ ? Maps.newHashMap()
+ : Maps.newHashMap(taskDescriptor.getProperties());
+ properties.put(TaskProperties.TASK_ATTEMPT_ID, Long.toString(next));
+ taskDescriptor.setProperties(properties);
+ }
+
+ private String currentAttemptId() {
+ Map<String, String> properties = taskDescriptor.getProperties();
+ return properties == null ? null : properties.get(TaskProperties.TASK_ATTEMPT_ID);
+ }
+
+ private static String reportedAttemptId(OptimizingTaskResult result) {
+ return result.getSummary() == null
+ ? null
+ : result.getSummary().get(TaskProperties.TASK_ATTEMPT_ID);
}
private void persistTaskRuntime() {
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/AMSServiceTestBase.java b/amoro-ams/src/test/java/org/apache/amoro/server/AMSServiceTestBase.java
index 27ad5e29a..b52b71c21 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/AMSServiceTestBase.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/AMSServiceTestBase.java
@@ -45,6 +45,9 @@ public abstract class AMSServiceTestBase extends AMSManagerTestBase {
configurations.set(AmoroManagementConf.OPTIMIZER_HB_TIMEOUT, Duration.ofMillis(800L));
configurations.set(
AmoroManagementConf.OPTIMIZER_TASK_EXECUTE_TIMEOUT, Duration.ofMillis(30000L));
+ // must stay above OPTIMIZER_POLLING_TIMEOUT (3s): a blocking pollTask waiting out its full
+ // timeout would otherwise pick up the task the keeper reset by ack-timeout in the meantime
+ configurations.set(AmoroManagementConf.OPTIMIZER_TASK_ACK_TIMEOUT, Duration.ofMillis(5000L));
configurations.set(
AmoroManagementConf.OPTIMIZER_GROUP_MIN_PARALLELISM_CHECK_INTERVAL,
Duration.ofMillis(10L));
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/TestDefaultOptimizingService.java b/amoro-ams/src/test/java/org/apache/amoro/server/TestDefaultOptimizingService.java
index 417150b39..365eee02d 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/TestDefaultOptimizingService.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/TestDefaultOptimizingService.java
@@ -35,8 +35,8 @@ import org.apache.amoro.catalog.BasicCatalogTestHelper;
import org.apache.amoro.catalog.CatalogTestHelper;
import org.apache.amoro.config.OptimizingConfig;
import org.apache.amoro.config.TableConfiguration;
-import org.apache.amoro.exception.IllegalTaskStateException;
import org.apache.amoro.exception.PluginRetryAuthException;
+import org.apache.amoro.exception.TaskRuntimeException;
import org.apache.amoro.io.MixedDataTestHelpers;
import org.apache.amoro.optimizing.RewriteFilesOutput;
import org.apache.amoro.optimizing.TableOptimizing;
@@ -302,9 +302,12 @@ public class TestDefaultOptimizingService extends AMSTableTestBase {
public void testAckAndCompleteTask() {
OptimizingTask task = optimizingService().pollTask(token, THREAD_ID);
Assertions.assertNotNull(task);
- Assertions.assertThrows(
- IllegalTaskStateException.class,
- () -> optimizingService().completeTask(token, buildOptimizingTaskResult(task.getTaskId())));
+ // Completing before ack is now treated as a stale response and absorbed silently (see
+ // TaskRuntime#complete): the result cannot be told apart from a stale completion for a task
+ // that
+ // was reset and re-scheduled to the same thread, so the task simply stays SCHEDULED.
+ optimizingService().completeTask(token, buildOptimizingTaskResult(task.getTaskId()));
+ assertTaskStatus(TaskRuntime.Status.SCHEDULED);
optimizingService().ackTask(token, THREAD_ID, task.getTaskId());
@@ -314,6 +317,29 @@ public class TestDefaultOptimizingService extends AMSTableTestBase {
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 (5s in tests). 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
+
+ // the optimizer stays alive (Toucher touches every 300ms), so waiting past the ack timeout hits
+ // the SCHEDULED + ackTimeout branch rather than the optimizer-expired branch: the keeper resets
+ // the task out from under the live optimizer
+ waitForTaskStatus(TaskRuntime.Status.PLANNED, 10000);
+
+ // the delayed ack arrives for the now-reset task -> rejected, exactly like the issue
+ Assertions.assertThrows(
+ TaskRuntimeException.class,
+ () -> optimizingService().ackTask(token, THREAD_ID, task.getTaskId()));
+ }
+
@Test
public void testExecuteTaskTimeOutAndRetry() throws InterruptedException {
OptimizingTask task = optimizingService().pollTask(token, THREAD_ID);
@@ -746,6 +772,19 @@ public class TestDefaultOptimizingService extends AMSTableTestBase {
optimizingService().listTasks(defaultResourceGroup().getName()).get(0).getStatus());
}
+ private void waitForTaskStatus(TaskRuntime.Status expectedStatus, long timeoutMs)
+ throws InterruptedException {
+ long deadline = System.currentTimeMillis() + timeoutMs;
+ while (System.currentTimeMillis() < deadline) {
+ if (expectedStatus
+ == optimizingService().listTasks(defaultResourceGroup().getName()).get(0).getStatus()) {
+ return;
+ }
+ Thread.sleep(100);
+ }
+ assertTaskStatus(expectedStatus);
+ }
+
private void assertTaskCompleted(TaskRuntime<?> taskRuntime) {
if (taskRuntime != null) {
Assertions.assertEquals(TaskRuntime.Status.SUCCESS, taskRuntime.getStatus());
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java
index 014189f64..772f15c01 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestOptimizingQueue.java
@@ -39,12 +39,14 @@ import org.apache.amoro.api.OptimizingTaskId;
import org.apache.amoro.api.OptimizingTaskResult;
import org.apache.amoro.catalog.BasicCatalogTestHelper;
import org.apache.amoro.catalog.CatalogTestHelper;
+import org.apache.amoro.exception.TaskRuntimeException;
import org.apache.amoro.io.MixedDataTestHelpers;
import org.apache.amoro.metrics.Gauge;
import org.apache.amoro.metrics.MetricKey;
import org.apache.amoro.metrics.MetricRegistry;
import org.apache.amoro.optimizing.RewriteFilesOutput;
import org.apache.amoro.optimizing.TableOptimizing;
+import org.apache.amoro.optimizing.TaskProperties;
import org.apache.amoro.process.ProcessStatus;
import org.apache.amoro.resource.ResourceGroup;
import org.apache.amoro.server.manager.MetricManager;
@@ -364,6 +366,119 @@ public class TestOptimizingQueue extends AMSTableTestBase {
queue.dispose();
}
+ // Issue #4235 fix: when the same optimizer thread polls again while one of its tasks is still
+ // ACKED, pollTask -> resetStaleTasksForThread resets that task. With more than one task in the
+ // process, the repoll schedules a *different* task, leaving the original PLANNED with token ==
+ // null. The optimizer's in-flight ack for the reset task is rejected on purpose so the optimizer
+ // skips executing this obsolete round; the task stays PLANNED to be re-polled by another thread.
+ @Test
+ public void testStaleAckAfterRepollIsRejected() {
+ DefaultTableRuntime tableRuntime = initTableWithPartitionedFiles();
+ OptimizingQueue queue = buildOptimizingGroupService(tableRuntime);
+
+ // 1. poll + ack task A -> ACKED (optimizer started executing it)
+ TaskRuntime<?> taskA = queue.pollTask(optimizerThread, MAX_POLLING_TIME);
+ Assert.assertNotNull(taskA);
+ queue.ackTask(taskA.getTaskId(), optimizerThread);
+ Assert.assertEquals(TaskRuntime.Status.ACKED, taskA.getStatus());
+
+ // 2. the same thread polls again; resetStaleTasksForThread resets the still-executing task A
+ // and
+ // the repoll schedules a different task B. Task A is left PLANNED with no token.
+ TaskRuntime<?> taskB = queue.pollTask(optimizerThread, MAX_POLLING_TIME);
+ Assert.assertNotNull(taskB);
+ Assert.assertNotEquals(taskA.getTaskId(), taskB.getTaskId());
+ Assert.assertEquals(TaskRuntime.Status.PLANNED, taskA.getStatus());
+ Assert.assertNull(taskA.getToken());
+
+ // 3. task A's in-flight ack is rejected so the optimizer skips this stale round (no duplicate
+ // execution). The exception still carries the message the optimizer client recognizes.
+ TaskRuntimeException e =
+ Assert.assertThrows(
+ TaskRuntimeException.class, () -> queue.ackTask(taskA.getTaskId(), optimizerThread));
+ Assert.assertTrue(e.getMessage().contains("has been reset or not yet scheduled"));
+
+ // task A remains PLANNED, still waiting to be re-polled and re-executed
+ Assert.assertEquals(TaskRuntime.Status.PLANNED, taskA.getStatus());
+ queue.dispose();
+ }
+
+ // Issue #4235 fix, single-task variant: the repoll re-schedules the *same* task (its token is
+ // restored), so the in-flight SUCCESS result for the previous run must be recognized as stale and
+ // ignored -- not blow up the SCHEDULED -> SUCCESS transition. The freshly re-scheduled round is
+ // left intact to be acked and completed normally.
+ @Test
+ public void testStaleCompleteAfterRepollIsIgnored() {
+ DefaultTableRuntime tableRuntime = initTableWithFiles();
+ OptimizingQueue queue = buildOptimizingGroupService(tableRuntime);
+
+ TaskRuntime<?> task = queue.pollTask(optimizerThread, MAX_POLLING_TIME);
+ Assert.assertNotNull(task);
+ queue.ackTask(task.getTaskId(), optimizerThread);
+ Assert.assertEquals(TaskRuntime.Status.ACKED, task.getStatus());
+
+ // same thread polls again -> resetStaleTasksForThread resets the executing task, then the
+ // single
+ // task is re-scheduled back to the same thread (now SCHEDULED again, awaiting its own ack).
+ TaskRuntime<?> repolled = queue.pollTask(optimizerThread, MAX_POLLING_TIME);
+ Assert.assertSame(task, repolled);
+ Assert.assertEquals(TaskRuntime.Status.SCHEDULED, task.getStatus());
+
+ // the previous run's SUCCESS result arrives; it is stale and must be ignored (no exception)
+ queue.completeTask(
+ optimizerThread,
+ buildOptimizingTaskResult(task.getTaskId(), optimizerThread.getThreadId()));
+
+ // the current re-scheduled round is untouched: still SCHEDULED, awaiting its own ack
+ Assert.assertEquals(TaskRuntime.Status.SCHEDULED, task.getStatus());
+ queue.dispose();
+ }
+
+ // The (token, threadId, status) checks alone cannot tell two attempts of the same task on the
+ // same optimizer thread apart: once the rescheduled round is ACKED again, a delayed completion
+ // of the previous attempt matches all three. Every schedule therefore stamps an attempt id into
+ // the task properties, and a completion echoing a different attempt id must be ignored.
+ @Test
+ public void testStaleCompleteFromPreviousAttemptIsIgnored() {
+ DefaultTableRuntime tableRuntime = initTableWithFiles();
+ OptimizingQueue queue = buildOptimizingGroupService(tableRuntime);
+
+ // 1. first attempt: poll + ack, the optimizer starts executing
+ TaskRuntime<?> task = queue.pollTask(optimizerThread, MAX_POLLING_TIME);
+ Assert.assertNotNull(task);
+ String firstAttempt =
+ task.extractProtocolTask().getProperties().get(TaskProperties.TASK_ATTEMPT_ID);
+ Assert.assertNotNull(firstAttempt);
+ queue.ackTask(task.getTaskId(), optimizerThread);
+
+ // 2. the same thread polls again: the task is reset and re-scheduled to the SAME thread with
+ // a new attempt id, then the second attempt is acked -> (token, threadId, ACKED) is now
+ // identical to what the first attempt's completion will carry
+ TaskRuntime<?> repolled = queue.pollTask(optimizerThread, MAX_POLLING_TIME);
+ Assert.assertSame(task, repolled);
+ String secondAttempt =
+ task.extractProtocolTask().getProperties().get(TaskProperties.TASK_ATTEMPT_ID);
+ Assert.assertNotEquals(firstAttempt, secondAttempt);
+ queue.ackTask(task.getTaskId(), optimizerThread);
+ Assert.assertEquals(TaskRuntime.Status.ACKED, task.getStatus());
+
+ // 3. the delayed completion of the FIRST attempt arrives; only the echoed attempt id can
+ // expose it as stale -> ignored, the second attempt keeps executing
+ OptimizingTaskResult staleResult =
+ buildOptimizingTaskResult(task.getTaskId(), optimizerThread.getThreadId());
+ staleResult.setSummary(ImmutableMap.of(TaskProperties.TASK_ATTEMPT_ID, firstAttempt));
+ queue.completeTask(optimizerThread, staleResult);
+ Assert.assertEquals(TaskRuntime.Status.ACKED, task.getStatus());
+
+ // 4. the second attempt's own completion (echoing the current attempt id) is accepted
+ OptimizingTaskResult currentResult =
+ buildOptimizingTaskResult(task.getTaskId(), optimizerThread.getThreadId());
+ currentResult.setSummary(ImmutableMap.of(TaskProperties.TASK_ATTEMPT_ID, secondAttempt));
+ queue.completeTask(optimizerThread, currentResult);
+ Assert.assertEquals(TaskRuntime.Status.SUCCESS, task.getStatus());
+ queue.dispose();
+ }
+
@Test
public void testCommitTask() {
DefaultTableRuntime tableRuntime = initTableWithFiles();
diff --git a/amoro-common/src/main/java/org/apache/amoro/process/StagedTaskDescriptor.java b/amoro-common/src/main/java/org/apache/amoro/process/StagedTaskDescriptor.java
index 638cb5971..0074e7a9f 100644
--- a/amoro-common/src/main/java/org/apache/amoro/process/StagedTaskDescriptor.java
+++ b/amoro-common/src/main/java/org/apache/amoro/process/StagedTaskDescriptor.java
@@ -85,6 +85,10 @@ public abstract class StagedTaskDescriptor<I, O, S> {
return properties;
}
+ public void setProperties(Map<String, String> properties) {
+ this.properties = properties;
+ }
+
public S getSummary() {
return summary;
}
diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/TaskProperties.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/TaskProperties.java
index 6a88da6b3..ede5467dc 100644
--- a/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/TaskProperties.java
+++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/optimizing/TaskProperties.java
@@ -31,6 +31,13 @@ public class TaskProperties {
public static final String PROCESS_ID = "process-id";
public static final String UNKNOWN_PROCESS_ID = "unknown";
+ /**
+ * Attempt id of a task execution round, stamped by AMS on every schedule and echoed back by the
+ * optimizer in the result summary, so AMS can tell a stale completion of a previous attempt apart
+ * from the current one.
+ */
+ public static final String TASK_ATTEMPT_ID = "task-attempt-id";
+
public static final String EXTEND_DISK_STORAGE =
OptimizerProperties.OPTIMIZER_EXTEND_DISK_STORAGE;
public static final boolean EXTEND_DISK_STORAGE_DEFAULT = false;
diff --git a/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/OptimizerExecutor.java b/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/OptimizerExecutor.java
index 2868cf083..371cc9069 100644
--- a/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/OptimizerExecutor.java
+++ b/amoro-optimizer/amoro-optimizer-common/src/main/java/org/apache/amoro/optimizer/common/OptimizerExecutor.java
@@ -156,6 +156,9 @@ public class OptimizerExecutor extends AbstractOptimizerOperator {
}
} finally {
if (result != null) {
+ // every completion passes through here, including error results built by executor
+ // subclasses (e.g. SparkOptimizerExecutor), so the attempt id is echoed exactly once
+ echoTaskAttemptId(ackTask, result);
completeTask(amsUrl, result);
}
}
@@ -328,6 +331,25 @@ public class OptimizerExecutor extends AbstractOptimizerOperator {
}
}
+ /**
+ * Echoes the attempt id received with the task back in the result summary, so AMS can tell a
+ * stale completion of a previous attempt apart from the current one. Tasks from older AMS
+ * versions carry no attempt id and the result is left untouched.
+ */
+ private static void echoTaskAttemptId(OptimizingTask task, OptimizingTaskResult result) {
+ String attemptId =
+ task.getProperties() == null
+ ? null
+ : task.getProperties().get(TaskProperties.TASK_ATTEMPT_ID);
+ if (attemptId == null) {
+ return;
+ }
+ Map<String, String> summary =
+ result.getSummary() == null ? Maps.newHashMap() : Maps.newHashMap(result.getSummary());
+ summary.put(TaskProperties.TASK_ATTEMPT_ID, attemptId);
+ result.setSummary(summary);
+ }
+
private static Map<String, String> fillTaskProperties(
OptimizerConfig config, OptimizingTask task) {
if (config.isCacheEnabled()) {
diff --git a/amoro-optimizer/amoro-optimizer-common/src/test/java/org/apache/amoro/optimizer/common/TestOptimizerExecutor.java b/amoro-optimizer/amoro-optimizer-common/src/test/java/org/apache/amoro/optimizer/common/TestOptimizerExecutor.java
index 866ad655a..3b4f9d933 100644
--- a/amoro-optimizer/amoro-optimizer-common/src/test/java/org/apache/amoro/optimizer/common/TestOptimizerExecutor.java
+++ b/amoro-optimizer/amoro-optimizer-common/src/test/java/org/apache/amoro/optimizer/common/TestOptimizerExecutor.java
@@ -158,6 +158,42 @@ public class TestOptimizerExecutor extends OptimizerTestBase {
1, TEST_AMS.getOptimizerHandler().getCompletedTasks().get(token).size());
}
+ @Test
+ public void testEchoTaskAttemptIdOnSuccess() throws InterruptedException, TException {
+ TEST_AMS.getOptimizerHandler().authenticate(new OptimizerRegisterInfo());
+ String token =
+ TEST_AMS.getOptimizerHandler().getRegisteredOptimizers().keySet().iterator().next();
+ OptimizingTask task = TestOptimizingInput.successInput(1).toTask(0, 0);
+ task.getProperties().put(TaskProperties.TASK_ATTEMPT_ID, "7");
+ TEST_AMS.getOptimizerHandler().offerTask(task);
+ optimizerExecutor.setToken(token);
+ TimeUnit.MILLISECONDS.sleep(OptimizerTestHelpers.CALL_AMS_INTERVAL * 2);
+ OptimizingTaskResult taskResult =
+ TEST_AMS.getOptimizerHandler().getCompletedTasks().get(token).get(0);
+ Assertions.assertNull(taskResult.getErrorMessage());
+ // TestOptimizingOutput.summary() is null, so the echo must create the summary map itself
+ Assertions.assertNotNull(taskResult.getSummary());
+ Assertions.assertEquals("7", taskResult.getSummary().get(TaskProperties.TASK_ATTEMPT_ID));
+ }
+
+ @Test
+ public void testEchoTaskAttemptIdOnFailure() throws InterruptedException, TException {
+ TEST_AMS.getOptimizerHandler().authenticate(new OptimizerRegisterInfo());
+ String token =
+ TEST_AMS.getOptimizerHandler().getRegisteredOptimizers().keySet().iterator().next();
+ OptimizingTask task = TestOptimizingInput.failedInput(1).toTask(0, 0);
+ task.getProperties().put(TaskProperties.TASK_ATTEMPT_ID, "7");
+ TEST_AMS.getOptimizerHandler().offerTask(task);
+ optimizerExecutor.setToken(token);
+ TimeUnit.MILLISECONDS.sleep(OptimizerTestHelpers.CALL_AMS_INTERVAL * 2);
+ OptimizingTaskResult taskResult =
+ TEST_AMS.getOptimizerHandler().getCompletedTasks().get(token).get(0);
+ // an error result reports the attempt id too, so a stale FAILED completion cannot poison the
+ // retry counting of the current attempt
+ Assertions.assertTrue(taskResult.getErrorMessage().contains(FAILED_TASK_MESSAGE));
+ Assertions.assertEquals("7", taskResult.getSummary().get(TaskProperties.TASK_ATTEMPT_ID));
+ }
+
@Test
public void testExecuteTaskFailed() throws InterruptedException, TException {
TEST_AMS.getOptimizerHandler().authenticate(new OptimizerRegisterInfo());