diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/process/executor/TableProcessExecutor.java b/amoro-ams/src/main/java/org/apache/amoro/server/process/executor/TableProcessExecutor.java index b16517e3a0..ef14c736d9 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/process/executor/TableProcessExecutor.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/process/executor/TableProcessExecutor.java @@ -24,6 +24,7 @@ import org.apache.amoro.process.TableProcess; import org.apache.amoro.process.TableProcessStore; import org.apache.amoro.server.persistence.PersistentBase; +import org.apache.amoro.shade.guava32.com.google.common.annotations.VisibleForTesting; import org.apache.amoro.shade.guava32.com.google.common.base.Strings; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -36,9 +37,11 @@ public class TableProcessExecutor extends PersistentBase implements Runnable { private static final Logger LOG = LoggerFactory.getLogger(TableProcessExecutor.class); private static final long DEFAULT_POLL_INTERVAL_MS = 5000L; + @VisibleForTesting static final int MAX_UNKNOWN_STATUS_POLLS = 3; public ExecuteEngine executeEngine; protected TableProcess tableProcess; private final TableProcessStore store; + private final long pollIntervalMs; private Runnable finishedCallback; /** @@ -49,9 +52,19 @@ public class TableProcessExecutor extends PersistentBase implements Runnable { */ public TableProcessExecutor( TableProcess tableProcess, TableProcessStore store, ExecuteEngine executeEngine) { + this(tableProcess, store, executeEngine, DEFAULT_POLL_INTERVAL_MS); + } + + @VisibleForTesting + TableProcessExecutor( + TableProcess tableProcess, + TableProcessStore store, + ExecuteEngine executeEngine, + long pollIntervalMs) { this.tableProcess = tableProcess; this.executeEngine = executeEngine; this.store = store; + this.pollIntervalMs = pollIntervalMs; } /** Submit or recover the process to engine, poll status and update store. */ @@ -86,29 +99,49 @@ public void run() { validateIdentifier(externalProcessIdentifier); status = executeEngine.getStatus(externalProcessIdentifier); + boolean submittedStatePersisted = false; + int consecutiveUnknownPolls = 0; - // If the engine returns UNKNOWN, the process was lost (e.g., AMS restart cleared - // LocalExecutionEngine's in-memory process map). Treat it as FAILED so the process - // is properly terminated and optionally retried, rather than being stuck in UNKNOWN. - if (status == ProcessStatus.UNKNOWN) { - LOG.warn( - "Table process {} got UNKNOWN status from engine (identifier={}), " - + "likely due to AMS restart or process loss, marking as FAILED", - store.getProcessId(), - externalProcessIdentifier); - status = ProcessStatus.FAILED; - message = "Process lost: engine returned UNKNOWN status"; - } else { - store.tryTransitState( - status, - ProcessEvent.SUBMIT_REQUESTED, - externalProcessIdentifier, - "Complete Submitted.", - tableProcess.getProcessParameters(), - tableProcess.getSummary()); - } - - while (isTableProcessExecuting(status)) { + while (status == ProcessStatus.UNKNOWN || isTableProcessExecuting(status)) { + if (status == ProcessStatus.UNKNOWN) { + if (++consecutiveUnknownPolls > MAX_UNKNOWN_STATUS_POLLS) { + LOG.error( + "Table process {} (identifier {}) stayed UNKNOWN for {} consecutive polls, " + + "cancelling best-effort", + store.getProcessId(), + externalProcessIdentifier, + consecutiveUnknownPolls); + try { + executeEngine.tryCancelTableProcess(tableProcess, externalProcessIdentifier); + } catch (Throwable cancelFailure) { + LOG.warn( + "Failed to cancel persistently UNKNOWN process {}", store.getProcessId(), cancelFailure); + } + status = ProcessStatus.FAILED; + message = + String.format( + "Engine reported UNKNOWN for %d consecutive polls", consecutiveUnknownPolls); + break; + } + LOG.warn( + "Table process {} (identifier {}) got UNKNOWN status, poll {} of {}, keep polling", + store.getProcessId(), + externalProcessIdentifier, + consecutiveUnknownPolls, + MAX_UNKNOWN_STATUS_POLLS); + } else { + consecutiveUnknownPolls = 0; + if (!submittedStatePersisted) { + store.tryTransitState( + status, + ProcessEvent.SUBMIT_REQUESTED, + externalProcessIdentifier, + "Complete Submitted.", + tableProcess.getProcessParameters(), + tableProcess.getSummary()); + submittedStatePersisted = true; + } + } if (isTableProcessCanceling(store.getStatus())) { LOG.info( "Table process {} with identifier {} may have been in canceling, exit submit process.", @@ -117,12 +150,22 @@ public void run() { return; } try { - Thread.sleep(DEFAULT_POLL_INTERVAL_MS); + Thread.sleep(pollIntervalMs); } catch (InterruptedException e) { throw e; } status = executeEngine.getStatus(externalProcessIdentifier); } + + if (!submittedStatePersisted && status != ProcessStatus.FAILED) { + store.tryTransitState( + status, + ProcessEvent.SUBMIT_REQUESTED, + externalProcessIdentifier, + "Complete Submitted.", + tableProcess.getProcessParameters(), + tableProcess.getSummary()); + } } catch (Throwable t) { if (t instanceof InterruptedException) { LOG.info( diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/process/executor/TestTableProcessExecutor.java b/amoro-ams/src/test/java/org/apache/amoro/server/process/executor/TestTableProcessExecutor.java index 69b693e8df..9460dbcf24 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/process/executor/TestTableProcessExecutor.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/process/executor/TestTableProcessExecutor.java @@ -36,13 +36,9 @@ /** Tests for {@link TableProcessExecutor}. */ public class TestTableProcessExecutor { - /** - * When a recovered process has a non-empty externalProcessIdentifier and the engine returns - * UNKNOWN (simulating AMS restart where LocalExecutionEngine's in-memory process map is cleared), - * the executor should mark the process as FAILED instead of leaving it stuck in UNKNOWN. - */ + /** A persistent UNKNOWN must eventually fail and trigger best-effort cancellation. */ @Test - public void testUnknownStatusFromEngineMarksAsFailed() { + public void testPersistentUnknownStatusMarksAsFailed() { // Mock store: simulate a recovered process with RUNNING status and a stale identifier TableProcessStore store = Mockito.mock(TableProcessStore.class); when(store.getStatus()).thenReturn(ProcessStatus.RUNNING); @@ -87,14 +83,55 @@ public void testUnknownStatusFromEngineMarksAsFailed() { return true; }); - TableProcessExecutor executor = new TableProcessExecutor(process, store, engine); + TableProcessExecutor executor = new TableProcessExecutor(process, store, engine, 1L); executor.run(); - // Verify the process was marked as FAILED, not left as UNKNOWN assertEquals( ProcessStatus.FAILED, finalStatus.get(), - "Process should be marked FAILED when engine returns UNKNOWN"); + "Process should fail after the configured UNKNOWN tolerance is exhausted"); + Mockito.verify(engine) + .tryCancelTableProcess(process, "stale-identifier-from-before-restart"); + } + + /** A single transient UNKNOWN must not abandon a job that becomes visible on the next poll. */ + @Test + public void testTransientUnknownStatusContinuesUntilSuccess() { + TableProcessStore store = Mockito.mock(TableProcessStore.class); + when(store.getStatus()).thenReturn(ProcessStatus.RUNNING); + when(store.getExternalProcessIdentifier()).thenReturn("recovering-identifier"); + when(store.getProcessId()).thenReturn(3L); + + ExecuteEngine engine = Mockito.mock(ExecuteEngine.class); + when(engine.getStatus("recovering-identifier")) + .thenReturn(ProcessStatus.UNKNOWN, ProcessStatus.SUCCESS); + + TableProcess process = Mockito.mock(TableProcess.class); + when(process.getProcessParameters()).thenReturn(java.util.Collections.emptyMap()); + when(process.getSummary()).thenReturn(java.util.Collections.emptyMap()); + + AtomicReference finalStatus = new AtomicReference<>(); + when(store.tryTransitState( + any(ProcessStatus.class), + any(ProcessEvent.class), + anyString(), + anyString(), + any(), + any())) + .thenAnswer( + invocation -> { + ProcessStatus status = invocation.getArgument(0); + if (status == ProcessStatus.SUCCESS || status == ProcessStatus.FAILED) { + finalStatus.set(status); + } + return true; + }); + + new TableProcessExecutor(process, store, engine, 1L).run(); + + assertEquals(ProcessStatus.SUCCESS, finalStatus.get()); + Mockito.verify(engine, Mockito.times(2)).getStatus("recovering-identifier"); + Mockito.verify(engine, Mockito.never()).tryCancelTableProcess(Mockito.any(), Mockito.anyString()); } /**