This is an automated email from the ASF dual-hosted git repository.
jinsongzhou 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 647827578 [AMORO-4113] Do not update snapshot markers on
completeProcess(false) (#4114)
647827578 is described below
commit 6478275784d526a399e294d3806770e6f57a2c08
Author: Jiwon Park <[email protected]>
AuthorDate: Mon Mar 9 17:40:41 2026 +0900
[AMORO-4113] Do not update snapshot markers on completeProcess(false)
(#4114)
When an optimizing process is closed without success (e.g., optimizer
group change, task retry exhaustion), completeProcess(false) was
unconditionally updating lastOptimizedSnapshotId and
lastOptimizedChangeSnapshotId. This caused SchedulingPolicy.isTablePending()
to return false, permanently preventing the table from being re-planned.
Fix by only updating snapshot markers when success is true.
Signed-off-by: Jiwon Park <[email protected]>
---
.../amoro/server/table/DefaultTableRuntime.java | 7 ++-
.../server/optimizing/TestOptimizingQueue.java | 73 ++++++++++++++++++++++
2 files changed, 78 insertions(+), 2 deletions(-)
diff --git
a/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java
b/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java
index baaebd278..002e7b0f2 100644
---
a/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java
+++
b/amoro-ams/src/main/java/org/apache/amoro/server/table/DefaultTableRuntime.java
@@ -385,8 +385,11 @@ public class DefaultTableRuntime extends
AbstractTableRuntime {
.updateState(
OPTIMIZING_STATE_KEY,
state -> {
-
state.setLastOptimizedSnapshotId(optimizingProcess.getTargetSnapshotId());
-
state.setLastOptimizedChangeSnapshotId(optimizingProcess.getTargetChangeSnapshotId());
+ if (success) {
+
state.setLastOptimizedSnapshotId(optimizingProcess.getTargetSnapshotId());
+ state.setLastOptimizedChangeSnapshotId(
+ optimizingProcess.getTargetChangeSnapshotId());
+ }
if (processType == OptimizingType.MINOR) {
state.setLastMinorOptimizingTime(optimizingProcess.getPlanTime());
} else if (processType == OptimizingType.MAJOR) {
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 ddf8fe985..bf26651a3 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
@@ -537,6 +537,79 @@ public class TestOptimizingQueue extends AMSTableTestBase {
queue.dispose();
}
+ @Test
+ public void testProcessCloseKeepsLastOptimizedSnapshotId() {
+ DefaultTableRuntime tableRuntime = initTableWithFiles();
+ long snapshotIdBeforePlanning = tableRuntime.getLastOptimizedSnapshotId();
+ long changeSnapshotIdBeforePlanning =
tableRuntime.getLastOptimizedChangeSnapshotId();
+
+ OptimizingQueue queue = buildOptimizingGroupService(tableRuntime);
+
+ // Poll task to trigger planning → process creation
+ TaskRuntime<?> task = queue.pollTask(optimizerThread, MAX_POLLING_TIME);
+ Assert.assertNotNull(task);
+
+ OptimizingProcess process = tableRuntime.getOptimizingProcess();
+ Assert.assertNotNull(process);
+ Assert.assertEquals(ProcessStatus.RUNNING, process.getStatus());
+
+ // Close process without success (simulates group change / forced
termination)
+ process.close(false);
+
+ // lastOptimizedSnapshotId and lastOptimizedChangeSnapshotId should NOT be
updated
+ Assert.assertEquals(snapshotIdBeforePlanning,
tableRuntime.getLastOptimizedSnapshotId());
+ Assert.assertEquals(
+ changeSnapshotIdBeforePlanning,
tableRuntime.getLastOptimizedChangeSnapshotId());
+ Assert.assertEquals(OptimizingStatus.IDLE,
tableRuntime.getOptimizingStatus());
+ Assert.assertNull(tableRuntime.getOptimizingProcess());
+ queue.dispose();
+ }
+
+ @Test
+ public void testTableRescheduledAfterProcessClose() {
+ DefaultTableRuntime tableRuntime = initTableWithFiles();
+
+ OptimizingQueue queue = buildOptimizingGroupService(tableRuntime);
+
+ // Poll task to trigger planning → process creation
+ TaskRuntime<?> task = queue.pollTask(optimizerThread, MAX_POLLING_TIME);
+ Assert.assertNotNull(task);
+
+ OptimizingProcess process = tableRuntime.getOptimizingProcess();
+ Assert.assertNotNull(process);
+
+ // Force close instead of commit (simulates group change)
+ process.close(false);
+ Assert.assertEquals(OptimizingStatus.IDLE,
tableRuntime.getOptimizingStatus());
+
+ // Verify snapshot marker was not updated — key condition for
isTablePending()
+ Assert.assertNotEquals(
+ tableRuntime.getLastOptimizedSnapshotId(),
tableRuntime.getCurrentSnapshotId());
+
+ // Simulate setPendingInput() transitioning IDLE → PENDING (done by
evaluator in production)
+ tableRuntime
+ .store()
+ .begin()
+ .updateStatusCode(code -> OptimizingStatus.PENDING.getCode())
+ .commit();
+ Assert.assertEquals(OptimizingStatus.PENDING,
tableRuntime.getOptimizingStatus());
+
+ // Set minPlanInterval to 0 so the table is immediately eligible for
re-planning
+ tableRuntime
+ .store()
+ .begin()
+ .updateTableConfig(
+ config ->
config.put(TableProperties.SELF_OPTIMIZING_MIN_PLAN_INTERVAL, "0"))
+ .commit();
+
+ // Build a new queue with the PENDING table — scheduler should NOT skip it
+ queue.dispose();
+ OptimizingQueue queue2 = buildOptimizingGroupService(tableRuntime);
+ TaskRuntime<?> task2 = queue2.pollTask(optimizerThread, MAX_POLLING_TIME);
+ Assert.assertNotNull(task2);
+ queue2.dispose();
+ }
+
protected DefaultTableRuntime initTableWithFiles() {
MixedTable mixedTable =
(MixedTable)
tableService().loadTable(serverTableIdentifier()).originalTable();
