This is an automated email from the ASF dual-hosted git repository.
zhoujinsong 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 acb9a49a4 [AMORO-4300] Recover active processes after table ownership
transfer (#4304)
acb9a49a4 is described below
commit acb9a49a493fc0708e186a7da6bac9634deef895
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]>
---
.../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 {
