This is an automated email from the ASF dual-hosted git repository. xxubai pushed a commit to branch 0.9.x in repository https://gitbox.apache.org/repos/asf/amoro.git
commit bb843c97cbffead377788fe1867d16d7498f6063 Author: Xu Bai <[email protected]> AuthorDate: Thu Jul 30 19:11:45 2026 +0800 [AMORO-4300] Recover active processes after table ownership transfer (#4304) [AMORO-4300][ams] Recover active processes after table ownership transfer Co-authored-by: ZhouJinsong <[email protected]> (cherry picked from commit acb9a49a493fc0708e186a7da6bac9634deef895) --- .../persistence/mapper/TableProcessMapper.java | 6 +- .../amoro/server/process/ProcessService.java | 80 +++++--- .../amoro/server/TestDefaultProcessService.java | 209 ++++++++++++++++++++- 3 files changed, 266 insertions(+), 29 deletions(-) diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/TableProcessMapper.java b/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/TableProcessMapper.java index ba21e01ff..21d8d14e3 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/TableProcessMapper.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/TableProcessMapper.java @@ -152,7 +152,9 @@ public interface TableProcessMapper { @Select( "SELECT process_id, table_id, external_process_identifier, status, process_type, process_stage, execution_engine, retry_number, " + "create_time, finish_time, fail_message, process_parameters, summary " - + "FROM table_process WHERE status in ('SUBMITTED', 'RUNNING')") + + "FROM table_process WHERE table_id in (#{tableIds::number[]}) " + + "AND status in ('SUBMITTED', 'RUNNING')") + @Lang(InListExtendedLanguageDriver.class) @ResultMap("tableProcessMap") - List<TableProcessMeta> selectAllActiveProcesses(); + List<TableProcessMeta> selectActiveProcesses(@Param("tableIds") Collection<Long> tableIds); } diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/process/ProcessService.java b/amoro-ams/src/main/java/org/apache/amoro/server/process/ProcessService.java index b166ada7a..957e3bf7b 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/process/ProcessService.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/process/ProcessService.java @@ -45,6 +45,7 @@ import java.util.ArrayList; import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.LinkedBlockingQueue; @@ -73,6 +74,9 @@ public class ProcessService extends PersistentBase { private final Map<ServerTableIdentifier, Map<Long, TableProcessHolder>> activeTableProcess = new ConcurrentHashMap<>(); + // Guards against concurrent recovery of the same processId (e.g. startup recovery racing with + // handleTableAdded). + private final Set<Long> recoveringProcessIds = ConcurrentHashMap.newKeySet(); public ProcessService(TableService tableService) { this(tableService, Collections.emptyList(), new ExecuteEngineManager()); @@ -207,8 +211,14 @@ public class ProcessService extends PersistentBase { Map<Long, TableRuntime> tableIdToRuntimes = tableRuntimes.stream() .collect(Collectors.toMap(t -> t.getTableIdentifier().getId(), t -> t)); + // InListExtendedLanguageDriver requires at least one value to produce a valid IN clause. + if (tableIdToRuntimes.isEmpty()) { + return; + } List<TableProcessMeta> activeProcesses = - getAs(TableProcessMapper.class, TableProcessMapper::selectAllActiveProcesses); + getAs( + TableProcessMapper.class, + mapper -> mapper.selectActiveProcesses(tableIdToRuntimes.keySet())); activeProcesses.forEach( processMeta -> { TableRuntime tableRuntime = tableIdToRuntimes.get(processMeta.getTableId()); @@ -237,7 +247,7 @@ public class ProcessService extends PersistentBase { /** * Recover a single persisted process record. Any failure is contained here: the offending record * is marked {@link ProcessStatus#FAILED} and skipped, so that one un-recoverable process record - * cannot abort the whole AMS startup (see AMORO-4223). The affected maintenance action will be + * cannot abort process recovery (see AMORO-4223). The affected maintenance action will be * re-scheduled by its periodic scheduler. * * @param tableRuntime table runtime @@ -248,32 +258,51 @@ public class ProcessService extends PersistentBase { TableRuntime tableRuntime, ActionCoordinatorScheduler scheduler, TableProcessMeta processMeta) { - DefaultTableProcessStore store = - new DefaultTableProcessStore( - processMeta.getProcessId(), - tableRuntime, - processMeta, - scheduler.getAction(), - processMeta.getRetryNumber()); + long processId = processMeta.getProcessId(); + if (!recoveringProcessIds.add(processId)) { + LOG.debug( + "Table process {} for table {} is already being recovered, skipping duplicate recovery.", + processId, + tableRuntime.getTableIdentifier()); + return; + } try { - TableProcess process = scheduler.recover(tableRuntime, store); - trackTableProcess(tableRuntime.getTableIdentifier(), store, process); - executeOrTraceProcess(store, process); - } catch (Throwable t) { - LOG.error( - "Failed to recover table process {} (action {}) for table {}, marking it FAILED " - + "and skipping so AMS can continue to start up.", - processMeta.getProcessId(), - scheduler.getAction(), - tableRuntime.getTableIdentifier(), - t); - markRecoverFailed(store, t); + if (getTableProcessInstances(tableRuntime.getTableIdentifier()).containsKey(processId)) { + LOG.debug( + "Table process {} for table {} is already tracked, skipping duplicate recovery.", + processId, + tableRuntime.getTableIdentifier()); + return; + } + DefaultTableProcessStore store = + new DefaultTableProcessStore( + processId, + tableRuntime, + processMeta, + scheduler.getAction(), + processMeta.getRetryNumber()); + try { + TableProcess process = scheduler.recover(tableRuntime, store); + trackTableProcess(tableRuntime.getTableIdentifier(), store, process); + executeOrTraceProcess(store, process); + } catch (Throwable t) { + LOG.error( + "Failed to recover table process {} (action {}) for table {}, marking it FAILED " + + "and continuing recovery.", + processId, + scheduler.getAction(), + tableRuntime.getTableIdentifier(), + t); + markRecoverFailed(store, t); + } + } finally { + recoveringProcessIds.remove(processId); } } /** * Best-effort mark an un-recoverable process as {@link ProcessStatus#FAILED} so it is not picked - * up again on the next AMS restart. Never throws. + * up again by a later recovery attempt. Never throws. * * @param store process store * @param cause the recovery failure @@ -284,13 +313,13 @@ public class ProcessService extends PersistentBase { ProcessStatus.FAILED, ProcessEvent.COMPLETE_FAILED, store.getExternalProcessIdentifier(), - "Failed to recover process on AMS startup: " + cause.getMessage(), + "Failed to recover process: " + cause.getMessage(), store.getProcessParameters(), store.getSummary()); } catch (Throwable t) { LOG.error( - "Failed to mark un-recoverable table process {} as FAILED; it may be retried on the " - + "next AMS restart.", + "Failed to mark un-recoverable table process {} as FAILED; a later recovery attempt " + + "may retry it.", store.getProcessId(), t); } @@ -635,6 +664,7 @@ public class ProcessService extends PersistentBase { */ @Override protected void handleTableAdded(AmoroTable<?> table, TableRuntime tableRuntime) { + recoverProcesses(Collections.singletonList(tableRuntime)); findSchedulersByFormat(tableRuntime.getFormat()) .forEach(s -> s.handleTableAdded(table, tableRuntime)); } diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/TestDefaultProcessService.java b/amoro-ams/src/test/java/org/apache/amoro/server/TestDefaultProcessService.java index c49bfe348..0fbcea418 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/TestDefaultProcessService.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/TestDefaultProcessService.java @@ -21,14 +21,19 @@ package org.apache.amoro.server; import org.apache.amoro.BasicTableTestHelper; import org.apache.amoro.ServerTableIdentifier; import org.apache.amoro.TableFormat; +import org.apache.amoro.TableRuntime; import org.apache.amoro.TableTestHelper; import org.apache.amoro.catalog.BasicCatalogTestHelper; import org.apache.amoro.catalog.CatalogTestHelper; import org.apache.amoro.process.ProcessStatus; +import org.apache.amoro.process.TableProcess; import org.apache.amoro.process.TableProcessStore; +import org.apache.amoro.server.persistence.PersistentBase; +import org.apache.amoro.server.persistence.mapper.TableProcessMapper; import org.apache.amoro.server.process.MockActionCoordinator; import org.apache.amoro.server.process.MockExecuteEngine; import org.apache.amoro.server.process.ProcessService; +import org.apache.amoro.server.process.TableProcessMeta; import org.apache.amoro.server.process.ThrowingRecoverActionCoordinator; import org.apache.amoro.server.table.AMSTableTestBase; import org.junit.After; @@ -41,7 +46,13 @@ import org.junit.runners.Parameterized; import java.util.ArrayList; import java.util.Collections; import java.util.Map; +import java.util.Optional; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BooleanSupplier; /** @@ -53,6 +64,7 @@ public class TestDefaultProcessService extends AMSTableTestBase { private static final long WAIT_TIMEOUT_MS = 60_000L; private static final long POLL_INTERVAL_MS = 3_000L; + private static final Persistence PERSISTENCE = new Persistence(); /** * Parameterization for catalog and table helpers. @@ -177,7 +189,7 @@ public class TestDefaultProcessService extends AMSTableTestBase { ProcessService.TableProcessHolder holder = getAnyActiveTableProcessHolder(); TableProcessStore store = holder.getStore(); - org.apache.amoro.TableRuntime tableRuntime = holder.getProcess().getTableRuntime(); + TableRuntime tableRuntime = holder.getProcess().getTableRuntime(); awaitEngineStatus(executeEngine, store.getExternalProcessIdentifier(), ProcessStatus.RUNNING); Assert.assertEquals(ProcessStatus.RUNNING, store.getStatus()); @@ -219,6 +231,121 @@ public class TestDefaultProcessService extends AMSTableTestBase { } } + /** Verify active processes are recovered when table ownership moves to this AMS. */ + @Test(timeout = 60_000) + public void testRecoverTableProcessWhenTableAdded() { + MockExecuteEngine executeEngine = getExecuteEngine(); + ExecutorService recoveryExecutor = Executors.newSingleThreadExecutor(); + BlockingRecoverActionCoordinator coordinator = + new BlockingRecoverActionCoordinator(executeEngine); + try { + // Start a process as the old table owner and capture its persisted identity. + createTable(); + awaitActiveInstances(executeEngine); + + ProcessService.TableProcessHolder originalHolder = getAnyActiveTableProcessHolder(); + TableProcessStore originalStore = originalHolder.getStore(); + TableRuntime tableRuntime = originalHolder.getProcess().getTableRuntime(); + long processId = originalStore.getProcessId(); + String originalExternalId = originalStore.getExternalProcessIdentifier(); + + // Simulate losing the old owner: stop its external process and wait until the local active + // process entry has been removed. + executeEngine.tryCancelTableProcess(originalHolder.getProcess(), originalExternalId); + awaitCondition( + () -> originalStore.getStatus() == ProcessStatus.CANCELED, + WAIT_TIMEOUT_MS, + POLL_INTERVAL_MS); + awaitCondition( + () -> + processServiceService() + .getTableProcessInstances(tableRuntime.getTableIdentifier()) + .isEmpty(), + WAIT_TIMEOUT_MS, + POLL_INTERVAL_MS); + Assert.assertFalse( + processServiceService() + .getTableProcessInstances(tableRuntime.getTableIdentifier()) + .containsValue(originalHolder)); + + // Recreate the database state observed after an abrupt owner loss. The process remains + // RUNNING, but its external identifier is unavailable to the new owner. + markProcessRunningWithoutExternalIdentifier(processId); + processServiceService().unInstallAllActionCoordinators(); + processServiceService().installActionCoordinator(coordinator); + + // Hold the first table-added recovery inside the coordinator, then deliver the same event + // again. The second event MUST see the atomic recovery reservation and return without + // recovering the process a second time. + Future<?> firstRecovery = + recoveryExecutor.submit( + () -> + processServiceService() + .getTableHandlerChain() + .fireTableAdded( + tableService().loadTable(tableRuntime.getTableIdentifier()), + tableRuntime)); + coordinator.awaitRecoveryStarted(); + + processServiceService() + .getTableHandlerChain() + .fireTableAdded( + tableService().loadTable(tableRuntime.getTableIdentifier()), tableRuntime); + Assert.assertEquals(1, coordinator.getRecoveryCount()); + + coordinator.releaseRecovery(); + firstRecovery.get(WAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS); + + // The new owner MUST track the same persisted process ID with a newly submitted external + // process. The old store must not remain in the active-process map. + awaitCondition( + () -> + processServiceService() + .getTableProcessInstances(tableRuntime.getTableIdentifier()) + .containsKey(processId), + WAIT_TIMEOUT_MS, + POLL_INTERVAL_MS); + + ProcessService.TableProcessHolder recoveredHolder = + processServiceService() + .getTableProcessInstances(tableRuntime.getTableIdentifier()) + .get(processId); + awaitCondition( + () -> + recoveredHolder.getStore().getStatus() == ProcessStatus.RUNNING + && !recoveredHolder.getStore().getExternalProcessIdentifier().isEmpty(), + WAIT_TIMEOUT_MS, + POLL_INTERVAL_MS); + + String recoveredExternalId = recoveredHolder.getStore().getExternalProcessIdentifier(); + Assert.assertNotEquals(originalExternalId, recoveredExternalId); + Assert.assertNotSame(originalStore, recoveredHolder.getStore()); + Assert.assertEquals( + 1, + processServiceService() + .getTableProcessInstances(tableRuntime.getTableIdentifier()) + .size()); + Assert.assertEquals(1, executeEngine.getActiveInstances().size()); + + // handleTableAdded also starts the periodic scheduler. Wait until it actually triggers and + // verify that the recovered RUNNING process prevents a second process from being submitted. + coordinator.awaitSchedulerTriggered(); + Assert.assertEquals(1, executeEngine.getActiveInstances().size()); + Assert.assertEquals( + 1, + processServiceService() + .getTableProcessInstances(tableRuntime.getTableIdentifier()) + .size()); + + dropTable(); + } catch (Throwable t) { + throw new RuntimeException(t); + } finally { + coordinator.releaseRecovery(); + recoveryExecutor.shutdownNow(); + } + } + /** * Verify that a single un-recoverable process record does not abort AMS startup: {@code * recoverProcesses} must not propagate the failure, the bad record is skipped and persisted as @@ -234,7 +361,7 @@ public class TestDefaultProcessService extends AMSTableTestBase { ProcessService.TableProcessHolder holder = getAnyActiveTableProcessHolder(); TableProcessStore store = holder.getStore(); - org.apache.amoro.TableRuntime tableRuntime = holder.getProcess().getTableRuntime(); + TableRuntime tableRuntime = holder.getProcess().getTableRuntime(); awaitEngineStatus(executeEngine, store.getExternalProcessIdentifier(), ProcessStatus.RUNNING); Assert.assertEquals(ProcessStatus.RUNNING, store.getStatus()); @@ -329,6 +456,84 @@ public class TestDefaultProcessService extends AMSTableTestBase { return getAnyActiveTableProcessHolder().getStore(); } + private void markProcessRunningWithoutExternalIdentifier(long processId) { + PERSISTENCE.markProcessRunningWithoutExternalIdentifier(processId); + } + + private static class Persistence extends PersistentBase { + private void markProcessRunningWithoutExternalIdentifier(long processId) { + doAs( + TableProcessMapper.class, + mapper -> { + TableProcessMeta meta = mapper.getProcessMeta(processId); + mapper.updateProcess( + meta.getTableId(), + processId, + "", + ProcessStatus.RUNNING, + meta.getProcessStage(), + meta.getRetryNumber(), + 0L, + "", + meta.getProcessParameters(), + meta.getSummary()); + }); + } + } + + private static class BlockingRecoverActionCoordinator extends MockActionCoordinator { + private final CountDownLatch recoveryStarted = new CountDownLatch(1); + private final CountDownLatch releaseRecovery = new CountDownLatch(1); + private final CountDownLatch schedulerTriggered = new CountDownLatch(1); + private final AtomicInteger recoveryCount = new AtomicInteger(); + + private BlockingRecoverActionCoordinator(MockExecuteEngine executeEngine) { + super(executeEngine); + } + + @Override + public TableProcess recoverTableProcess( + TableRuntime tableRuntime, TableProcessStore processStore) { + recoveryCount.incrementAndGet(); + recoveryStarted.countDown(); + try { + if (!releaseRecovery.await(WAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS)) { + throw new AssertionError("Timed out waiting to release process recovery"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException("Interrupted while waiting to recover process", e); + } + return super.recoverTableProcess(tableRuntime, processStore); + } + + @Override + public Optional<TableProcess> trigger(TableRuntime tableRuntime) { + schedulerTriggered.countDown(); + return super.trigger(tableRuntime); + } + + private void awaitRecoveryStarted() throws InterruptedException { + if (!recoveryStarted.await(WAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS)) { + throw new AssertionError("Process recovery did not start"); + } + } + + private void releaseRecovery() { + releaseRecovery.countDown(); + } + + private void awaitSchedulerTriggered() throws InterruptedException { + if (!schedulerTriggered.await(WAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS)) { + throw new AssertionError("Table scheduler did not trigger"); + } + } + + private int getRecoveryCount() { + return recoveryCount.get(); + } + } + /** Wait until the given externalProcessIdentifier reaches the specified status. */ private void awaitEngineStatus(MockExecuteEngine engine, String externalId, ProcessStatus status) throws InterruptedException {
