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 <[email protected]>
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 <[email protected]>
* [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 <[email protected]>
* [AMORO-4235] Apply spotless formatting
Signed-off-by: Jiwon Park <[email protected]>
* [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 <[email protected]>
* [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 <[email protected]>
---------
Signed-off-by: Jiwon Park <[email protected]>
Co-authored-by: ConradJam <[email protected]>
---
.../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());
