You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@amoro.apache.org by xu...@apache.org on 2026/07/31 07:28:48 UTC
(amoro) branch 0.9.x updated (30580a3ae -> f46f57d94)
This is an automated email from the ASF dual-hosted git repository.
xxubai pushed a change to branch 0.9.x
in repository https://gitbox.apache.org/repos/asf/amoro.git
from 30580a3ae Merge remote-tracking branch 'origin/0.9.0-rc4' into 0.9.x
new 48735a8e3 [hotfix] Fix S3 credentials not vended when loading Lance table (#4279)
new 25b2a8809 [AMORO-4274] Remove Roaring64Bitmap.isEmpty() check in CombinedDelete… (#4275)
new dbd5f6ca6 [AMORO-4281]Fix cleanning process trigger issue for mixed format tables (#4282)
new 4e09b201c [AMORO-4126][ams] Default waitAppCompletion=false for Spark optimizer on Kubernetes (#4289)
new 5f618fb62 [AMORO-4277] imporove expire_snapshots performance (#4278)
new 0e4c6df32 [AMORO-4286]Change process status to failed when recover from AMS restarting (#4288)
new 6c4872763 [Hotfix] Pin all external actions to the specific git hash (#4294)
new 7a757db36 [Hotfix] Change the docker action version to match ASF policy (#4299)
new fed7feba8 [AMORO-4297][AMS] Persist failure reasons in task quota history (#4298)
new bb843c97c [AMORO-4300] Recover active processes after table ownership transfer (#4304)
new f46f57d94 [AMORO-4301][ams] Fix Kubernetes executor label prefix (#4302)
The 11 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.github/workflows/docker-images.yml | 30 +--
.../server/manager/SparkOptimizerContainer.java | 11 +-
.../amoro/server/optimizing/TaskRuntime.java | 1 +
.../persistence/mapper/TableProcessMapper.java | 6 +-
.../amoro/server/process/ProcessService.java | 223 +++++++++++++++++----
.../process/executor/TableProcessExecutor.java | 28 ++-
.../amoro/server/utils/RollingFileCleaner.java | 138 -------------
.../amoro/server/TestDefaultProcessService.java | 209 ++++++++++++++++++-
.../manager/TestSparkOptimizerContainer.java | 95 +++++++++
.../TestTaskRuntime.java} | 24 +--
.../process/TestProcessServiceMultiFormat.java | 188 +++++++++++++++++
.../process/executor/TestTableProcessExecutor.java | 152 ++++++++++++++
.../formats/iceberg/utils/RollingFileCleaner.java | 41 ++--
.../amoro/io/reader/CombinedDeleteFilter.java | 2 +-
.../java/org/apache/amoro/utils/TableFileUtil.java | 18 +-
.../iceberg/utils}/TestRollingFileCleaner.java | 3 +-
.../formats/lance/LanceDirectoryV1Catalog.java | 1 +
17 files changed, 930 insertions(+), 240 deletions(-)
delete mode 100644 amoro-ams/src/main/java/org/apache/amoro/server/utils/RollingFileCleaner.java
copy amoro-ams/src/test/java/org/apache/amoro/server/{util/TestCompressUtil.java => optimizing/TestTaskRuntime.java} (63%)
create mode 100644 amoro-ams/src/test/java/org/apache/amoro/server/process/TestProcessServiceMultiFormat.java
create mode 100644 amoro-ams/src/test/java/org/apache/amoro/server/process/executor/TestTableProcessExecutor.java
rename {amoro-ams/src/test/java/org/apache/amoro/server/util => amoro-format-iceberg/src/test/java/org/apache/amoro/formats/iceberg/utils}/TestRollingFileCleaner.java (95%)
(amoro) 10/11: [AMORO-4300] Recover active processes after table ownership transfer (#4304)
Posted by xu...@apache.org.
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 <to...@gmail.com>
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 <zh...@163.com>
(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 {
(amoro) 03/11: [AMORO-4281]Fix cleanning process trigger issue for mixed format tables (#4282)
Posted by xu...@apache.org.
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 dbd5f6ca6bf3ffdce1e1b0446e1f63f0d1bf8377
Author: ZhouJinsong <zh...@163.com>
AuthorDate: Wed Jul 22 19:28:00 2026 +0800
[AMORO-4281]Fix cleanning process trigger issue for mixed format tables (#4282)
* Fix expire process trigger issue
* Add some method to reuse some codes
* Fix some issues of comments
---------
Co-authored-by: baiyangtx <xi...@163.com>
(cherry picked from commit 0704a1ee92df82bd923074d6714cb7612b45d964)
---
.../amoro/server/process/ProcessService.java | 143 +++++++++++++---
.../process/TestProcessServiceMultiFormat.java | 188 +++++++++++++++++++++
2 files changed, 312 insertions(+), 19 deletions(-)
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 68446db8d..b166ada7a 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
@@ -41,10 +41,12 @@ import org.apache.amoro.shade.guava32.com.google.common.annotations.VisibleForTe
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -59,7 +61,7 @@ public class ProcessService extends PersistentBase {
private static final Logger LOG = LoggerFactory.getLogger(ProcessService.class);
private final TableService tableService;
- private final Map<String, ActionCoordinatorScheduler> actionCoordinators =
+ private final Map<String, List<ActionCoordinatorScheduler>> actionCoordinators =
new ConcurrentHashMap<>();
private final Map<String, ExecuteEngine> executeEngines = new ConcurrentHashMap<>();
@@ -120,13 +122,70 @@ public class ProcessService extends PersistentBase {
activeTableProcess.clear();
}
+ /**
+ * Find the scheduler that supports the given table format for the specified action.
+ *
+ * @param actionName action name
+ * @param format table format
+ * @return matching scheduler, or null if not found
+ */
+ private ActionCoordinatorScheduler findScheduler(String actionName, TableFormat format) {
+ List<ActionCoordinatorScheduler> schedulers = actionCoordinators.get(actionName);
+ if (schedulers == null) {
+ return null;
+ }
+ for (ActionCoordinatorScheduler scheduler : schedulers) {
+ if (scheduler.formatSupported(format)) {
+ return scheduler;
+ }
+ }
+ return null;
+ }
+
+ /**
+ * Find all schedulers that support the given table format across all actions.
+ *
+ * @param format table format
+ * @return list of matching schedulers
+ */
+ private List<ActionCoordinatorScheduler> findSchedulersByFormat(TableFormat format) {
+ List<ActionCoordinatorScheduler> result = new ArrayList<>();
+ for (List<ActionCoordinatorScheduler> schedulers : actionCoordinators.values()) {
+ for (ActionCoordinatorScheduler scheduler : schedulers) {
+ if (scheduler.formatSupported(format)) {
+ result.add(scheduler);
+ }
+ }
+ }
+ return result;
+ }
+
private void initialize(List<TableRuntime> tableRuntimes) {
LOG.info("Initializing process service");
// Pre-configured coordinators built from TableRuntimeFactory / ProcessFactory
for (ActionCoordinator actionCoordinator : actionCoordinatorList) {
- actionCoordinators.put(
- actionCoordinator.action().getName(),
- new ActionCoordinatorScheduler(actionCoordinator, tableService, ProcessService.this));
+ ActionCoordinatorScheduler scheduler =
+ new ActionCoordinatorScheduler(actionCoordinator, tableService, ProcessService.this);
+ List<ActionCoordinatorScheduler> schedulers =
+ actionCoordinators.computeIfAbsent(
+ actionCoordinator.action().getName(), k -> new CopyOnWriteArrayList<>());
+ // Validate (action, format) uniqueness: warn if a scheduler with overlapping format support
+ // already exists for the same action
+ for (ActionCoordinatorScheduler existing : schedulers) {
+ for (TableFormat format : TableFormat.values()) {
+ if (existing.getCoordinator().formatSupported(format)
+ && actionCoordinator.formatSupported(format)) {
+ LOG.warn(
+ "Duplicate ActionCoordinator for action {} and format {}: existing={}, new={}. "
+ + "The later one will shadow the former in findScheduler.",
+ actionCoordinator.action().getName(),
+ format,
+ existing.getCoordinator().name(),
+ actionCoordinator.name());
+ }
+ }
+ }
+ schedulers.add(scheduler);
}
executeEngineManager
.installedPlugins()
@@ -135,7 +194,7 @@ public class ProcessService extends PersistentBase {
executeEngines.put(executeEngine.name(), executeEngine);
});
recoverProcesses(tableRuntimes);
- actionCoordinators.values().forEach(s -> s.initialize(tableRuntimes));
+ actionCoordinators.values().forEach(list -> list.forEach(s -> s.initialize(tableRuntimes)));
}
/**
@@ -153,10 +212,24 @@ public class ProcessService extends PersistentBase {
activeProcesses.forEach(
processMeta -> {
TableRuntime tableRuntime = tableIdToRuntimes.get(processMeta.getTableId());
- ActionCoordinatorScheduler scheduler =
- actionCoordinators.get(processMeta.getProcessType());
- if (tableRuntime != null && scheduler != null) {
- recoverProcess(tableRuntime, scheduler, processMeta);
+ if (tableRuntime != null) {
+ ActionCoordinatorScheduler scheduler =
+ findScheduler(processMeta.getProcessType(), tableRuntime.getFormat());
+ if (scheduler != null) {
+ recoverProcess(tableRuntime, scheduler, processMeta);
+ } else {
+ LOG.warn(
+ "No ActionCoordinatorScheduler found for process {} (type={}, format={}), "
+ + "skipping recovery. The process record will remain in its current state.",
+ processMeta.getProcessId(),
+ processMeta.getProcessType(),
+ tableRuntime.getFormat());
+ }
+ } else {
+ LOG.warn(
+ "Table runtime not found for process {} (tableId={}), skipping recovery.",
+ processMeta.getProcessId(),
+ processMeta.getTableId());
}
});
}
@@ -250,8 +323,11 @@ public class ProcessService extends PersistentBase {
TableProcessExecutor executor = new TableProcessExecutor(process, store, executeEngine);
executor.onProcessFinished(
() -> {
- ActionCoordinatorScheduler scheduler =
- actionCoordinators.get(store.getAction().getName());
+ ActionCoordinatorScheduler scheduler = null;
+ if (process.getTableRuntime() != null) {
+ scheduler =
+ findScheduler(store.getAction().getName(), process.getTableRuntime().getFormat());
+ }
if (scheduler != null
&& store.getStatus() == ProcessStatus.FAILED
&& store.getRetryNumber() < ActionCoordinatorScheduler.PROCESS_MAX_RETRY_NUMBER
@@ -265,6 +341,14 @@ public class ProcessService extends PersistentBase {
process.getSummary());
executeOrTraceProcess(store, process);
} else {
+ if (store.getStatus() == ProcessStatus.FAILED && process.getTableRuntime() != null) {
+ LOG.info(
+ "FAILED process {} will not be retried: scheduler={}, retryNumber={}, maxRetry={}",
+ store.getProcessId(),
+ scheduler != null ? "found" : "not found",
+ store.getRetryNumber(),
+ ActionCoordinatorScheduler.PROCESS_MAX_RETRY_NUMBER);
+ }
untrackTableProcessInstance(
process.getTableRuntime().getTableIdentifier(), store.getProcessId());
}
@@ -389,7 +473,7 @@ public class ProcessService extends PersistentBase {
* @return coordinators map
*/
@VisibleForTesting
- public Map<String, ActionCoordinatorScheduler> getActionCoordinators() {
+ public Map<String, List<ActionCoordinatorScheduler>> getActionCoordinators() {
return actionCoordinators;
}
@@ -467,9 +551,26 @@ public class ProcessService extends PersistentBase {
@VisibleForTesting
public void installActionCoordinator(ActionCoordinator actionCoordinator) {
- this.actionCoordinators.put(
- actionCoordinator.action().getName(),
- new ActionCoordinatorScheduler(actionCoordinator, tableService, ProcessService.this));
+ ActionCoordinatorScheduler scheduler =
+ new ActionCoordinatorScheduler(actionCoordinator, tableService, ProcessService.this);
+ List<ActionCoordinatorScheduler> schedulers =
+ this.actionCoordinators.computeIfAbsent(
+ actionCoordinator.action().getName(), k -> new CopyOnWriteArrayList<>());
+ // Validate (action, format) uniqueness
+ for (ActionCoordinatorScheduler existing : schedulers) {
+ for (TableFormat format : TableFormat.values()) {
+ if (existing.getCoordinator().formatSupported(format)
+ && actionCoordinator.formatSupported(format)) {
+ LOG.warn(
+ "Duplicate ActionCoordinator for action {} and format {}: existing={}, new={}.",
+ actionCoordinator.action().getName(),
+ format,
+ existing.getCoordinator().name(),
+ actionCoordinator.name());
+ }
+ }
+ }
+ schedulers.add(scheduler);
}
@VisibleForTesting
@@ -509,7 +610,8 @@ public class ProcessService extends PersistentBase {
*/
@Override
protected void handleStatusChanged(TableRuntime tableRuntime, OptimizingStatus originalStatus) {
- actionCoordinators.values().forEach(s -> s.handleStatusChanged(tableRuntime, originalStatus));
+ findSchedulersByFormat(tableRuntime.getFormat())
+ .forEach(s -> s.handleStatusChanged(tableRuntime, originalStatus));
}
/**
@@ -521,7 +623,8 @@ public class ProcessService extends PersistentBase {
@Override
protected void handleConfigChanged(
TableRuntime tableRuntime, TableConfiguration originalConfig) {
- actionCoordinators.values().forEach(s -> s.handleConfigChanged(tableRuntime, originalConfig));
+ findSchedulersByFormat(tableRuntime.getFormat())
+ .forEach(s -> s.handleConfigChanged(tableRuntime, originalConfig));
}
/**
@@ -532,7 +635,8 @@ public class ProcessService extends PersistentBase {
*/
@Override
protected void handleTableAdded(AmoroTable<?> table, TableRuntime tableRuntime) {
- actionCoordinators.values().forEach(s -> s.handleTableAdded(table, tableRuntime));
+ findSchedulersByFormat(tableRuntime.getFormat())
+ .forEach(s -> s.handleTableAdded(table, tableRuntime));
}
/**
@@ -542,7 +646,8 @@ public class ProcessService extends PersistentBase {
*/
@Override
protected void handleTableRemoved(TableRuntime tableRuntime) {
- actionCoordinators.values().forEach(s -> s.handleTableRemoved(tableRuntime));
+ findSchedulersByFormat(tableRuntime.getFormat())
+ .forEach(s -> s.handleTableRemoved(tableRuntime));
List<TableProcessHolder> processes =
getTableProcessInstances(tableRuntime.getTableIdentifier()).values().stream()
.collect(Collectors.toList());
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/process/TestProcessServiceMultiFormat.java b/amoro-ams/src/test/java/org/apache/amoro/server/process/TestProcessServiceMultiFormat.java
new file mode 100644
index 000000000..d0eab0992
--- /dev/null
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/process/TestProcessServiceMultiFormat.java
@@ -0,0 +1,188 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.amoro.server.process;
+
+import org.apache.amoro.Action;
+import org.apache.amoro.TableFormat;
+import org.apache.amoro.TableRuntime;
+import org.apache.amoro.process.ActionCoordinator;
+import org.apache.amoro.process.TableProcess;
+import org.apache.amoro.process.TableProcessStore;
+import org.apache.amoro.server.process.ProcessService.ExecuteEngineManager;
+import org.apache.amoro.server.table.TableService;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+/**
+ * Tests for {@link ProcessService} multi-format ActionCoordinator isolation. Ensures that
+ * coordinators sharing the same action name but targeting different formats are not silently
+ * overwritten (regression test for AMORO-4281).
+ */
+public class TestProcessServiceMultiFormat {
+
+ private static final Action TEST_ACTION = Action.register("test-expire-snapshots");
+
+ /** A coordinator that only supports a specific format. */
+ static class FormatSpecificCoordinator implements ActionCoordinator {
+ private final TableFormat format;
+ private final String name;
+
+ FormatSpecificCoordinator(TableFormat format) {
+ this.format = format;
+ this.name = "coordinator-" + format.name();
+ }
+
+ @Override
+ public boolean formatSupported(TableFormat format) {
+ return this.format.equals(format);
+ }
+
+ @Override
+ public int parallelism() {
+ return 1;
+ }
+
+ @Override
+ public Action action() {
+ return TEST_ACTION;
+ }
+
+ @Override
+ public long getNextExecutingTime(TableRuntime tableRuntime) {
+ return 1000L;
+ }
+
+ @Override
+ public boolean enabled(TableRuntime tableRuntime) {
+ return formatSupported(tableRuntime.getFormat());
+ }
+
+ @Override
+ public long getExecutorDelay() {
+ return 1000L;
+ }
+
+ @Override
+ public Optional<TableProcess> trigger(TableRuntime tableRuntime) {
+ return Optional.empty();
+ }
+
+ @Override
+ public TableProcess recoverTableProcess(
+ TableRuntime tableRuntime, TableProcessStore processStore) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public void open(Map<String, String> properties) {}
+
+ @Override
+ public void close() {}
+
+ @Override
+ public String name() {
+ return name;
+ }
+
+ TableFormat supportedFormat() {
+ return format;
+ }
+ }
+
+ /**
+ * Two coordinators with the same action name but different formats should both be registered
+ * without one overwriting the other. This is the core regression test for AMORO-4281.
+ */
+ @Test
+ public void testMultipleCoordinatorsWithSameActionDifferentFormats() {
+ FormatSpecificCoordinator icebergCoordinator =
+ new FormatSpecificCoordinator(TableFormat.ICEBERG);
+ FormatSpecificCoordinator paimonCoordinator = new FormatSpecificCoordinator(TableFormat.PAIMON);
+
+ TableService tableService = Mockito.mock(TableService.class);
+ ProcessService processService =
+ new ProcessService(tableService, Collections.emptyList(), new ExecuteEngineManager());
+
+ processService.installActionCoordinator(icebergCoordinator);
+ processService.installActionCoordinator(paimonCoordinator);
+
+ Map<String, List<ActionCoordinatorScheduler>> coordinators =
+ processService.getActionCoordinators();
+ Assertions.assertEquals(1, coordinators.size(), "Should have one action entry");
+ List<ActionCoordinatorScheduler> schedulers = coordinators.get(TEST_ACTION.getName());
+ Assertions.assertNotNull(schedulers, "Schedulers list should not be null");
+ Assertions.assertEquals(2, schedulers.size(), "Both coordinators should be registered");
+
+ // Verify both formats are represented
+ boolean hasIceberg =
+ schedulers.stream().anyMatch(s -> s.getCoordinator().formatSupported(TableFormat.ICEBERG));
+ boolean hasPaimon =
+ schedulers.stream().anyMatch(s -> s.getCoordinator().formatSupported(TableFormat.PAIMON));
+ Assertions.assertTrue(hasIceberg, "ICEBERG coordinator should be present");
+ Assertions.assertTrue(hasPaimon, "PAIMON coordinator should be present");
+ }
+
+ /**
+ * Installing coordinators via installActionCoordinator should also not overwrite existing ones
+ * with the same action name.
+ */
+ @Test
+ public void testInstallActionCoordinatorDoesNotOverwrite() {
+ FormatSpecificCoordinator icebergCoordinator =
+ new FormatSpecificCoordinator(TableFormat.ICEBERG);
+ FormatSpecificCoordinator paimonCoordinator = new FormatSpecificCoordinator(TableFormat.PAIMON);
+
+ TableService tableService = Mockito.mock(TableService.class);
+ ProcessService processService =
+ new ProcessService(tableService, Collections.emptyList(), new ExecuteEngineManager());
+
+ processService.installActionCoordinator(icebergCoordinator);
+ processService.installActionCoordinator(paimonCoordinator);
+
+ Map<String, List<ActionCoordinatorScheduler>> coordinators =
+ processService.getActionCoordinators();
+ List<ActionCoordinatorScheduler> schedulers = coordinators.get(TEST_ACTION.getName());
+ Assertions.assertNotNull(schedulers);
+ Assertions.assertEquals(2, schedulers.size(), "Both coordinators should be registered");
+ }
+
+ /**
+ * The getActionCoordinators return type should be Map<String, List<ActionCoordinatorScheduler>>.
+ */
+ @Test
+ public void testGetActionCoordinatorsReturnType() {
+ FormatSpecificCoordinator coordinator = new FormatSpecificCoordinator(TableFormat.ICEBERG);
+ TableService tableService = Mockito.mock(TableService.class);
+ ProcessService processService =
+ new ProcessService(tableService, Collections.emptyList(), new ExecuteEngineManager());
+
+ processService.installActionCoordinator(coordinator);
+
+ Map<String, List<ActionCoordinatorScheduler>> coordinators =
+ processService.getActionCoordinators();
+ Assertions.assertNotNull(coordinators);
+ Assertions.assertTrue(coordinators.containsKey(TEST_ACTION.getName()));
+ }
+}
(amoro) 08/11: [Hotfix] Change the docker action version to match ASF policy (#4299)
Posted by xu...@apache.org.
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 7a757db36b1360b4bdc021154b4c6a65d4f0c2ad
Author: ZhouJinsong <zh...@163.com>
AuthorDate: Wed Jul 29 11:58:32 2026 +0800
[Hotfix] Change the docker action version to match ASF policy (#4299)
Change the docker action version to match ASF policy
(cherry picked from commit ec28554a2eaab45c402d6c5a1a1c14ab36d769a1)
---
.github/workflows/docker-images.yml | 30 +++++++++++++++---------------
1 file changed, 15 insertions(+), 15 deletions(-)
diff --git a/.github/workflows/docker-images.yml b/.github/workflows/docker-images.yml
index 2e4e946ce..3632bc64f 100644
--- a/.github/workflows/docker-images.yml
+++ b/.github/workflows/docker-images.yml
@@ -51,13 +51,13 @@ jobs:
cache: maven
check-latest: false
- name: Set up QEMU
- uses: docker/setup-qemu-action@c7c53464625b32c7a7e944ae62b3e17d2b600130 # v3.7.0
+ uses: docker/setup-qemu-action@96fe6ef7f33517b61c61be40b68a1882f3264fb8 # v4.2.0
- name: Set up Docker Buildx
- uses: docker/setup-buildx-action@f7ce87c1d6bead3e36075b2ce75da1f6cc28aaca # v3.9.0
+ uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4.2.0
- name: Set up Docker tags
- uses: docker/metadata-action@318604b99e75e41977312d83839a89be02ca4893 # v5.9.0
+ uses: docker/metadata-action@dc802804100637a589fabce1cb79ff13a1411302 # v6.2.0
id: meta
with:
flavor: |
@@ -76,7 +76,7 @@ jobs:
run: echo '${{ steps.meta.outputs.tags }}'
- name: Login to Docker Hub
- uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # v3.7.0
+ uses: docker/login-action@371161bbe7024a29a25c5e19bfcbc0804fe9ad2c # v4.5.2
with:
username: ${{ secrets.DOCKERHUB_USER }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
@@ -90,7 +90,7 @@ jobs:
run: ./mvnw clean package -pl 'dist' -am -e ${MVN_HADOOP} -DskipTests -B -ntp -Psupport-all-formats -Pno-extended-disk-storage -Pno-plugin-bin
- name: Build and Push Amoro Docker Image
- uses: docker/build-push-action@4f58ea79222b3b9dc2c8bbdd6debcef730109a75 # v6.9.0
+ uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0
with:
context: .
push: true
@@ -115,13 +115,13 @@ jobs:
cache: maven
check-latest: false
- name: Set up QEMU
- uses: docker/setup-qemu-action@c7c53464625b32c7a7e944ae62b3e17d2b600130 # v3.7.0
+ uses: docker/setup-qemu-action@96fe6ef7f33517b61c61be40b68a1882f3264fb8 # v4.2.0
- name: Set up Docker Buildx
- uses: docker/setup-buildx-action@f7ce87c1d6bead3e36075b2ce75da1f6cc28aaca # v3.9.0
+ uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4.2.0
- name: Set up Docker tags
- uses: docker/metadata-action@318604b99e75e41977312d83839a89be02ca4893 # v5.9.0
+ uses: docker/metadata-action@dc802804100637a589fabce1cb79ff13a1411302 # v6.2.0
id: meta
with:
flavor: |
@@ -141,7 +141,7 @@ jobs:
run: echo '${{ steps.meta.outputs.tags }}'
- name: Login to Docker Hub
- uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # v3.7.0
+ uses: docker/login-action@371161bbe7024a29a25c5e19bfcbc0804fe9ad2c # v4.5.2
with:
username: ${{ secrets.DOCKERHUB_USER }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
@@ -167,7 +167,7 @@ jobs:
run: ./mvnw clean package -pl 'amoro-optimizer/amoro-optimizer-flink' -am -e ${OPTIMIZER_FLINK} -DskipTests -B -ntp
- name: Build and Push Flink Optimizer Docker Image
- uses: docker/build-push-action@4f58ea79222b3b9dc2c8bbdd6debcef730109a75 # v6.9.0
+ uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0
env:
AMORO_VERSION: ${{ steps.version.outputs.AMORO_VERSION }}
with:
@@ -198,13 +198,13 @@ jobs:
cache: maven
check-latest: false
- name: Set up QEMU
- uses: docker/setup-qemu-action@c7c53464625b32c7a7e944ae62b3e17d2b600130 # v3.7.0
+ uses: docker/setup-qemu-action@96fe6ef7f33517b61c61be40b68a1882f3264fb8 # v4.2.0
- name: Set up Docker Buildx
- uses: docker/setup-buildx-action@f7ce87c1d6bead3e36075b2ce75da1f6cc28aaca # v3.9.0
+ uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c # v4.2.0
- name: Set up Docker tags
- uses: docker/metadata-action@318604b99e75e41977312d83839a89be02ca4893 # v5.9.0
+ uses: docker/metadata-action@dc802804100637a589fabce1cb79ff13a1411302 # v6.2.0
id: meta
with:
flavor: |
@@ -222,7 +222,7 @@ jobs:
run: echo '${{ steps.meta.outputs.tags }}'
- name: Login to Docker Hub
- uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # v3.7.0
+ uses: docker/login-action@371161bbe7024a29a25c5e19bfcbc0804fe9ad2c # v4.5.2
with:
username: ${{ secrets.DOCKERHUB_USER }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
@@ -248,7 +248,7 @@ jobs:
run: ./mvnw clean package -pl 'amoro-optimizer/amoro-optimizer-spark' -am -e ${SPARK_VERSION} -DskipTests -B -ntp
- name: Build and Push Spark Optimizer Docker Image
- uses: docker/build-push-action@4f58ea79222b3b9dc2c8bbdd6debcef730109a75 # v6.9.0
+ uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a # v7.3.0
env:
AMORO_VERSION: ${{ steps.version.outputs.AMORO_VERSION }}
SPARK_MAJOR_VERSION: ${{ steps.versions.outputs.SPARK_MAJOR_VERSION }}
(amoro) 05/11: [AMORO-4277] imporove expire_snapshots performance (#4278)
Posted by xu...@apache.org.
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 5f618fb629aca220665a1a2eac85326889615def
Author: Wang Tao <wt...@gmail.com>
AuthorDate: Fri Jul 24 15:45:39 2026 +0800
[AMORO-4277] imporove expire_snapshots performance (#4278)
* optimize expire snapshots
fix: logical error
remove some comments
* remove duplicate class and move test case to amoro-format-iceberg
* Update amoro-format-iceberg/src/main/java/org/apache/amoro/utils/TableFileUtil.java
Co-authored-by: Xu Bai <to...@gmail.com>
* fix review comment
---------
Co-authored-by: Xu Bai <to...@gmail.com>
(cherry picked from commit 7be0710b3e134f889ce1f970604428663c0ca97b)
---
.../amoro/server/utils/RollingFileCleaner.java | 138 ---------------------
.../formats/iceberg/utils/RollingFileCleaner.java | 41 +++---
.../java/org/apache/amoro/utils/TableFileUtil.java | 18 ++-
.../iceberg/utils}/TestRollingFileCleaner.java | 3 +-
4 files changed, 44 insertions(+), 156 deletions(-)
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/utils/RollingFileCleaner.java b/amoro-ams/src/main/java/org/apache/amoro/server/utils/RollingFileCleaner.java
deleted file mode 100644
index 821bf6661..000000000
--- a/amoro-ams/src/main/java/org/apache/amoro/server/utils/RollingFileCleaner.java
+++ /dev/null
@@ -1,138 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.amoro.server.utils;
-
-import org.apache.amoro.io.AuthenticatedFileIO;
-import org.apache.amoro.shade.guava32.com.google.common.collect.Sets;
-import org.apache.amoro.utils.TableFileUtil;
-import org.apache.hadoop.fs.Path;
-import org.apache.iceberg.io.BulkDeletionFailureException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.net.URI;
-import java.util.Set;
-import java.util.concurrent.atomic.AtomicInteger;
-
-public class RollingFileCleaner {
- private final Set<String> collectedFiles = Sets.newConcurrentHashSet();
- private final Set<String> excludeFiles;
- private final Set<String> parentDirectories = Sets.newConcurrentHashSet();
-
- private static final int CLEANED_FILE_GROUP_SIZE = 1_000;
-
- private final AtomicInteger fileCounter = new AtomicInteger(0);
- private final AtomicInteger cleanedFileCounter = new AtomicInteger(0);
-
- private final AuthenticatedFileIO fileIO;
-
- private static final Logger LOG = LoggerFactory.getLogger(RollingFileCleaner.class);
-
- public RollingFileCleaner(AuthenticatedFileIO fileIO, Set<String> excludeFiles) {
- this.fileIO = fileIO;
- this.excludeFiles =
- excludeFiles != null
- ? Sets.newConcurrentHashSet(excludeFiles)
- : Sets.newConcurrentHashSet();
- }
-
- public void addFile(String filePath) {
- if (isFileExcluded(filePath)) {
- LOG.debug("File {} is excluded from cleaning", filePath);
- return;
- }
-
- collectedFiles.add(filePath);
- String parentDir = TableFileUtil.getParent(filePath);
- parentDirectories.add(parentDir);
- int currentCount = fileCounter.incrementAndGet();
-
- if (currentCount % CLEANED_FILE_GROUP_SIZE == 0) {
- doCleanFiles();
- }
- }
-
- private boolean isFileExcluded(String filePath) {
- if (excludeFiles.isEmpty()) {
- return false;
- }
-
- if (excludeFiles.contains(filePath)) {
- return true;
- }
-
- String uriPath = URI.create(filePath).getPath();
- if (excludeFiles.contains(uriPath)) {
- return true;
- }
-
- String parentPath = new Path(uriPath).getParent().toString();
- return excludeFiles.contains(parentPath);
- }
-
- private void doCleanFiles() {
- if (collectedFiles.isEmpty()) {
- return;
- }
-
- if (fileIO.supportBulkOperations()) {
- try {
- fileIO.asBulkFileIO().deleteFiles(collectedFiles);
- cleanedFileCounter.addAndGet(collectedFiles.size());
- } catch (BulkDeletionFailureException e) {
- LOG.warn("Failed to delete {} expired files in bulk", e.numberFailedObjects());
- }
- } else {
- for (String filePath : collectedFiles) {
- try {
- fileIO.deleteFile(filePath);
- cleanedFileCounter.incrementAndGet();
- } catch (Exception e) {
- LOG.warn("Failed to delete expired file: {}", filePath, e);
- }
- }
- }
- // Try to delete empty parent directories
- for (String parentDir : parentDirectories) {
- TableFileUtil.deleteEmptyDirectory(fileIO, parentDir, excludeFiles);
- }
- parentDirectories.clear();
-
- LOG.debug("Cleaned expired a file group, total files: {}", collectedFiles.size());
-
- collectedFiles.clear();
- }
-
- public int fileCount() {
- return fileCounter.get();
- }
-
- public int cleanedFileCount() {
- return cleanedFileCounter.get();
- }
-
- public void clear() {
- if (!collectedFiles.isEmpty()) {
- doCleanFiles();
- }
-
- collectedFiles.clear();
- excludeFiles.clear();
- }
-}
diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/utils/RollingFileCleaner.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/utils/RollingFileCleaner.java
index 8d2e8ac06..3fcdcb5b2 100644
--- a/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/utils/RollingFileCleaner.java
+++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/utils/RollingFileCleaner.java
@@ -27,8 +27,11 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.net.URI;
+import java.util.Comparator;
+import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.Collectors;
/** Rolling file cleaner for Iceberg table maintenance operations. */
public class RollingFileCleaner {
@@ -110,27 +113,34 @@ public class RollingFileCleaner {
}
}
}
- // Try to delete empty parent directories. Skipped automatically by
- // TableFileUtil.deleteEmptyDirectory when the underlying FileIO is an object store.
- if (fileIO.supportFileSystemOperations()) {
- for (String parentDir : parentDirectories) {
- try {
- TableFileUtil.deleteEmptyDirectory(fileIO, parentDir, excludeFiles);
- } catch (Exception e) {
- LOG.warn("Failed to delete empty parent directory: {}", parentDir, e);
- }
- }
- }
LOG.debug("Cleaned expired a file group, total files: {}", collectedFiles.size());
} finally {
- // Always drop both buffers so a failure in directory cleanup does not cause the
- // already-deleted files to be re-submitted on the next batch.
- parentDirectories.clear();
collectedFiles.clear();
}
}
+ private void doCleanParentDirectory() {
+ // Try to delete empty parent directories. Skipped automatically by
+ // TableFileUtil.deleteEmptyDirectory when the underlying FileIO is an object store.
+ if (fileIO.supportFileSystemOperations()) {
+ List<String> parentDirectoriesSorted =
+ parentDirectories.stream()
+ .sorted(Comparator.comparingInt(String::length).reversed())
+ .collect(Collectors.toList());
+ for (String parentDir : parentDirectoriesSorted) {
+ try {
+ TableFileUtil.deleteEmptyDirectory(fileIO, parentDir, excludeFiles, parentDirectories);
+ } catch (Exception e) {
+ LOG.warn("Failed to delete empty parent directory: {}", parentDir, e);
+ }
+ }
+ }
+ // Always drop both buffers so a failure in directory cleanup does not cause the
+ // already-deleted files to be re-submitted on the next batch.
+ parentDirectories.clear();
+ }
+
public int fileCount() {
return fileCounter.get();
}
@@ -143,6 +153,9 @@ public class RollingFileCleaner {
if (!collectedFiles.isEmpty()) {
doCleanFiles();
}
+ if (!parentDirectories.isEmpty()) {
+ doCleanParentDirectory();
+ }
collectedFiles.clear();
excludeFiles.clear();
diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/utils/TableFileUtil.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/utils/TableFileUtil.java
index 1bbe17d01..5ec207467 100644
--- a/amoro-format-iceberg/src/main/java/org/apache/amoro/utils/TableFileUtil.java
+++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/utils/TableFileUtil.java
@@ -28,6 +28,7 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.net.URI;
+import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicInteger;
@@ -75,15 +76,25 @@ public class TableFileUtil {
return filePath.substring(0, lastSlash);
}
+ public static void deleteEmptyDirectory(
+ AuthenticatedFileIO io, String directoryPath, Set<String> exclude) {
+ deleteEmptyDirectory(io, directoryPath, exclude, new HashSet<>());
+ }
+
/**
* Try to recursiveDelete the empty directory
*
* @param io mixed-format file io
* @param directoryPath directory location
* @param exclude the directory will not be deleted
+ * @param directoriesToBeDeleted: all the directories that need to be deleted directories that
+ * need to be deleted
*/
public static void deleteEmptyDirectory(
- AuthenticatedFileIO io, String directoryPath, Set<String> exclude) {
+ AuthenticatedFileIO io,
+ String directoryPath,
+ Set<String> exclude,
+ Set<String> directoriesToBeDeleted) {
if (directoryPath == null || directoryPath.isEmpty()) {
return;
}
@@ -108,7 +119,10 @@ public class TableFileUtil {
if (io.asFileSystemIO().isEmptyDirectory(directoryPath)) {
io.asFileSystemIO().deletePrefix(directoryPath);
LOG.debug("success delete empty directory {}", directoryPath);
- deleteEmptyDirectory(io, parent, exclude);
+ // for parent must be deleted after the sub-directory
+ if (directoriesToBeDeleted == null || !directoriesToBeDeleted.contains(parent)) {
+ deleteEmptyDirectory(io, parent, exclude);
+ }
}
}
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/util/TestRollingFileCleaner.java b/amoro-format-iceberg/src/test/java/org/apache/amoro/formats/iceberg/utils/TestRollingFileCleaner.java
similarity index 95%
rename from amoro-ams/src/test/java/org/apache/amoro/server/util/TestRollingFileCleaner.java
rename to amoro-format-iceberg/src/test/java/org/apache/amoro/formats/iceberg/utils/TestRollingFileCleaner.java
index 5931f72f3..e3461b588 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/util/TestRollingFileCleaner.java
+++ b/amoro-format-iceberg/src/test/java/org/apache/amoro/formats/iceberg/utils/TestRollingFileCleaner.java
@@ -16,11 +16,10 @@
* limitations under the License.
*/
-package org.apache.amoro.server.util;
+package org.apache.amoro.formats.iceberg.utils;
import org.apache.amoro.io.AuthenticatedFileIO;
import org.apache.amoro.io.AuthenticatedFileIOAdapter;
-import org.apache.amoro.server.utils.RollingFileCleaner;
import org.apache.amoro.shade.guava32.com.google.common.collect.Sets;
import org.apache.iceberg.inmemory.InMemoryFileIO;
import org.junit.jupiter.api.Assertions;
(amoro) 11/11: [AMORO-4301][ams] Fix Kubernetes executor label prefix (#4302)
Posted by xu...@apache.org.
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 f46f57d9423538520d7cafb05a398fa33acd5dbc
Author: csurong <cs...@gmail.com>
AuthorDate: Thu Jul 30 21:39:30 2026 +0800
[AMORO-4301][ams] Fix Kubernetes executor label prefix (#4302)
Use Spark's executor label prefix for optimizer executor pods and add regression coverage for both driver and executor labels.
Co-authored-by: caosurong <su...@thinkingdata.cn>
Co-authored-by: ZhouJinsong <zh...@163.com>
(cherry picked from commit efbedc396fa178abe39ec24071f91f24163d8c9a)
---
.../server/manager/SparkOptimizerContainer.java | 3 ++-
.../manager/TestSparkOptimizerContainer.java | 22 ++++++++++++++++++++++
2 files changed, 24 insertions(+), 1 deletion(-)
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/manager/SparkOptimizerContainer.java b/amoro-ams/src/main/java/org/apache/amoro/server/manager/SparkOptimizerContainer.java
index d453798fe..775ecfdbd 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/manager/SparkOptimizerContainer.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/manager/SparkOptimizerContainer.java
@@ -340,7 +340,8 @@ public class SparkOptimizerContainer extends AbstractOptimizerContainer {
public static final String KUBERNETES_NAMESPACE = "spark.kubernetes.namespace";
public static final String KUBERNETES_SUBMISSION_WAIT_APP_COMPLETION =
"spark.kubernetes.submission.waitAppCompletion";
- public static final String KUBERNETES_EXECUTOR_LABEL_PREFIX = "spark.kubernetes.driver.label.";
+ public static final String KUBERNETES_EXECUTOR_LABEL_PREFIX =
+ "spark.kubernetes.executor.label.";
public static final String KUBERNETES_DRIVER_LABEL_PREFIX = "spark.kubernetes.driver.label.";
public static final String KUBERNETES_DRA_ENABLED = "spark.dynamicAllocation.enabled";
public static final String KUBERNETES_DRA_MAX_EXECUTORS =
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/manager/TestSparkOptimizerContainer.java b/amoro-ams/src/test/java/org/apache/amoro/server/manager/TestSparkOptimizerContainer.java
index 5a432feb8..c109848aa 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/manager/TestSparkOptimizerContainer.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/manager/TestSparkOptimizerContainer.java
@@ -75,6 +75,17 @@ public class TestSparkOptimizerContainer {
+ "=false"));
}
+ @Test
+ public void testKubernetesAddsLabelsToDriverAndExecutorPods() {
+ SparkOptimizerContainer container = createContainer("k8s://https://127.0.0.1:6443");
+ Resource resource = createResource(Maps.newHashMap());
+
+ String startupArgs = container.buildOptimizerStartupArgsString(resource);
+
+ assertPodLabels(startupArgs, "spark.kubernetes.driver.label.", resource);
+ assertPodLabels(startupArgs, "spark.kubernetes.executor.label.", resource);
+ }
+
@Test
public void testKubernetesWaitAppCompletionCanBeOverridden() {
SparkOptimizerContainer container = createContainer("k8s://https://127.0.0.1:6443");
@@ -131,4 +142,15 @@ public class TestSparkOptimizerContainer {
.setProperties(properties)
.build();
}
+
+ private void assertPodLabels(String startupArgs, String labelPrefix, Resource resource) {
+ Assert.assertTrue(
+ startupArgs.contains(
+ "--conf " + labelPrefix + "optimizer-group=" + resource.getGroupName()));
+ Assert.assertTrue(
+ startupArgs.contains(
+ "--conf " + labelPrefix + "optimizer-implementation=spark-native-kubernetes"));
+ Assert.assertTrue(
+ startupArgs.contains("--conf " + labelPrefix + "optimizer-id=" + resource.getResourceId()));
+ }
}
(amoro) 01/11: [hotfix] Fix S3 credentials not vended when loading Lance table (#4279)
Posted by xu...@apache.org.
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 48735a8e32950a72dbcc143d9b9cb06be977d607
Author: WenLingzhang <12...@qq.com>
AuthorDate: Tue Jul 21 15:50:08 2026 +0800
[hotfix] Fix S3 credentials not vended when loading Lance table (#4279)
fixup bug
Co-authored-by: 张文领 <zh...@chinatelecom.cn>
(cherry picked from commit cbd7d65652226f9f7b259924b3d2164b5ea8d7a5)
---
.../java/org/apache/amoro/formats/lance/LanceDirectoryV1Catalog.java | 1 +
1 file changed, 1 insertion(+)
diff --git a/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceDirectoryV1Catalog.java b/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceDirectoryV1Catalog.java
index e522c3711..cbb3ce5a2 100755
--- a/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceDirectoryV1Catalog.java
+++ b/amoro-format-lance/src/main/java/org/apache/amoro/formats/lance/LanceDirectoryV1Catalog.java
@@ -81,6 +81,7 @@ public class LanceDirectoryV1Catalog implements FormatCatalog {
Preconditions.checkArgument(
root != null && !root.isEmpty(), "Warehouse must be set in catalogProperties.");
this.namespaceProperties.put("manifest_enabled", "false");
+ this.namespaceProperties.put("vend_input_storage_options", "true");
this.namespaceProperties.put("root", root);
if (storageAccessKey != null) {
this.namespaceProperties.put(STORAGE_ACCESS_KEY_ID, storageAccessKey);
(amoro) 06/11: [AMORO-4286]Change process status to failed when recover from AMS restarting (#4288)
Posted by xu...@apache.org.
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 0e4c6df32bc0b9a9ff0418d5f63c7e79d8c74dec
Author: ZhouJinsong <zh...@163.com>
AuthorDate: Mon Jul 27 17:40:28 2026 +0800
[AMORO-4286]Change process status to failed when recover from AMS restarting (#4288)
* Change process status to failed when recover from unknown status
* Fix the checkstyle errors
(cherry picked from commit 436a3173b992cf049112b0ea47cb83f38a2afc6f)
---
.../process/executor/TableProcessExecutor.java | 28 +++-
.../process/executor/TestTableProcessExecutor.java | 152 +++++++++++++++++++++
2 files changed, 173 insertions(+), 7 deletions(-)
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 83cd66a3f..b16517e3a 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
@@ -86,13 +86,27 @@ public class TableProcessExecutor extends PersistentBase implements Runnable {
validateIdentifier(externalProcessIdentifier);
status = executeEngine.getStatus(externalProcessIdentifier);
- store.tryTransitState(
- status,
- ProcessEvent.SUBMIT_REQUESTED,
- externalProcessIdentifier,
- "Complete Submitted.",
- tableProcess.getProcessParameters(),
- tableProcess.getSummary());
+
+ // 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)) {
if (isTableProcessCanceling(store.getStatus())) {
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
new file mode 100644
index 000000000..69b693e8d
--- /dev/null
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/process/executor/TestTableProcessExecutor.java
@@ -0,0 +1,152 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.amoro.server.process.executor;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.when;
+
+import org.apache.amoro.process.ExecuteEngine;
+import org.apache.amoro.process.ProcessEvent;
+import org.apache.amoro.process.ProcessStatus;
+import org.apache.amoro.process.TableProcess;
+import org.apache.amoro.process.TableProcessStore;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+/** 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.
+ */
+ @Test
+ public void testUnknownStatusFromEngineMarksAsFailed() {
+ // 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);
+ when(store.getExternalProcessIdentifier()).thenReturn("stale-identifier-from-before-restart");
+ when(store.getProcessId()).thenReturn(1L);
+ when(store.tryTransitState(
+ any(ProcessStatus.class),
+ any(ProcessEvent.class),
+ anyString(),
+ anyString(),
+ any(),
+ any()))
+ .thenReturn(true);
+
+ // Mock engine: returns UNKNOWN for the stale identifier (process lost after restart)
+ ExecuteEngine engine = Mockito.mock(ExecuteEngine.class);
+ when(engine.getStatus("stale-identifier-from-before-restart"))
+ .thenReturn(ProcessStatus.UNKNOWN);
+
+ // Mock table process
+ TableProcess process = Mockito.mock(TableProcess.class);
+ when(process.getProcessParameters()).thenReturn(java.util.Collections.emptyMap());
+ when(process.getSummary()).thenReturn(java.util.Collections.emptyMap());
+
+ // Track the final status transit
+ AtomicReference<ProcessStatus> finalStatus = new AtomicReference<>();
+ when(store.tryTransitState(
+ any(ProcessStatus.class),
+ any(ProcessEvent.class),
+ anyString(),
+ anyString(),
+ any(),
+ any()))
+ .thenAnswer(
+ invocation -> {
+ ProcessStatus status = invocation.getArgument(0);
+ // Capture the FAILED status transit (not the SUBMIT_REQUESTED which is skipped)
+ if (status == ProcessStatus.FAILED) {
+ finalStatus.set(status);
+ when(store.getStatus()).thenReturn(ProcessStatus.FAILED);
+ }
+ return true;
+ });
+
+ TableProcessExecutor executor = new TableProcessExecutor(process, store, engine);
+ 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");
+ }
+
+ /**
+ * When the engine returns a normal terminal status (SUCCESS), the executor should transit to
+ * SUCCESS normally. This is a regression test to ensure the UNKNOWN fix doesn't break the normal
+ * path.
+ */
+ @Test
+ public void testNormalSuccessPathNotAffected() {
+ TableProcessStore store = Mockito.mock(TableProcessStore.class);
+ when(store.getStatus()).thenReturn(ProcessStatus.RUNNING);
+ when(store.getExternalProcessIdentifier()).thenReturn("valid-identifier");
+ when(store.getProcessId()).thenReturn(2L);
+ when(store.tryTransitState(
+ any(ProcessStatus.class),
+ any(ProcessEvent.class),
+ anyString(),
+ anyString(),
+ any(),
+ any()))
+ .thenReturn(true);
+
+ ExecuteEngine engine = Mockito.mock(ExecuteEngine.class);
+ when(engine.getStatus("valid-identifier")).thenReturn(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<ProcessStatus> 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.RUNNING) {
+ finalStatus.set(status);
+ }
+ return true;
+ });
+
+ TableProcessExecutor executor = new TableProcessExecutor(process, store, engine);
+ executor.run();
+
+ assertEquals(
+ ProcessStatus.SUCCESS,
+ finalStatus.get(),
+ "Process should complete with SUCCESS on normal path");
+ }
+}
(amoro) 09/11: [AMORO-4297][AMS] Persist failure reasons in task quota history (#4298)
Posted by xu...@apache.org.
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 fed7feba8adffcc158a3f5b99fc6b22f1ada7831
Author: Jiwon Park <jp...@outlook.kr>
AuthorDate: Thu Jul 30 15:21:11 2026 +0900
[AMORO-4297][AMS] Persist failure reasons in task quota history (#4298)
fix: persist task quota failure reasons
Signed-off-by: Jiwon Park <jp...@outlook.kr>
(cherry picked from commit b7bf0d39de9bb085446f1b06da2cb1810ac5cb26)
---
.../amoro/server/optimizing/TaskRuntime.java | 1 +
.../amoro/server/optimizing/TestTaskRuntime.java | 38 ++++++++++++++++++++++
2 files changed, 39 insertions(+)
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 fc7e67877..dc7fc99d5 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
@@ -408,6 +408,7 @@ public class TaskRuntime<T extends StagedTaskDescriptor<?, ?, ?>> extends Stated
this.taskId = task.getTaskId().getTaskId();
this.tableId = task.getTableId();
this.retryNum = task.getRetry();
+ this.failReason = task.getFailReason();
}
public long getStartTime() {
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestTaskRuntime.java b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestTaskRuntime.java
new file mode 100644
index 000000000..aee7e122e
--- /dev/null
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/optimizing/TestTaskRuntime.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.amoro.server.optimizing;
+
+import org.apache.amoro.api.OptimizingTaskId;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+public class TestTaskRuntime {
+
+ @Test
+ void taskQuotaCopiesFailReason() {
+ TaskRuntime<?> task = Mockito.mock(TaskRuntime.class);
+ Mockito.when(task.getTaskId()).thenReturn(new OptimizingTaskId(1L, 1));
+ Mockito.when(task.getFailReason()).thenReturn("task failed");
+
+ TaskRuntime.TaskQuota quota = new TaskRuntime.TaskQuota(task);
+
+ Assertions.assertEquals("task failed", quota.getFailReason());
+ }
+}
(amoro) 02/11: [AMORO-4274] Remove Roaring64Bitmap.isEmpty() check in CombinedDelete… (#4275)
Posted by xu...@apache.org.
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 25b2a8809f5ca7976246e734a68b916d387db7e0
Author: byungnam <lb...@gmail.com>
AuthorDate: Wed Jul 22 16:06:48 2026 +0900
[AMORO-4274] Remove Roaring64Bitmap.isEmpty() check in CombinedDelete… (#4275)
[AMORO-4274] Remove Roaring64Bitmap.isEmpty() check in CombinedDeleteFilter
positionMap is populated only via computeIfAbsent(...).add(...) while reading
position-delete files, so any stored value is a non-empty Roaring64Bitmap.
positionMap.get(...) therefore returns null or a non-empty bitmap, never an
empty one, making the isEmpty() branch unreachable.
Roaring64Bitmap.isEmpty() is backed by getLongCardinality(), which traverses
all containers, so calling it once per scanned record wastes CPU on the hot
path. Drop the check and keep only the null guard; behavior is unchanged.
Close #4274
(cherry picked from commit 0cb625072686a58d32af00939d919569a7185586)
---
.../src/main/java/org/apache/amoro/io/reader/CombinedDeleteFilter.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/amoro-format-iceberg/src/main/java/org/apache/amoro/io/reader/CombinedDeleteFilter.java b/amoro-format-iceberg/src/main/java/org/apache/amoro/io/reader/CombinedDeleteFilter.java
index d184ffcde..6440f0ff2 100644
--- a/amoro-format-iceberg/src/main/java/org/apache/amoro/io/reader/CombinedDeleteFilter.java
+++ b/amoro-format-iceberg/src/main/java/org/apache/amoro/io/reader/CombinedDeleteFilter.java
@@ -415,7 +415,7 @@ public abstract class CombinedDeleteFilter<T extends StructLike> {
return structLikeForDelete -> {
Roaring64Bitmap posSet = positionMap.get(structLikeForDelete.filePath());
- if (posSet == null || posSet.isEmpty()) {
+ if (posSet == null) {
return false;
}
return posSet.contains(structLikeForDelete.getPosition());
(amoro) 04/11: [AMORO-4126][ams] Default waitAppCompletion=false for Spark optimizer on Kubernetes (#4289)
Posted by xu...@apache.org.
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 4e09b201c47c35f1b2c57408d37ec7642229c0fa
Author: csurong <cs...@gmail.com>
AuthorDate: Fri Jul 24 11:49:22 2026 +0800
[AMORO-4126][ams] Default waitAppCompletion=false for Spark optimizer on Kubernetes (#4289)
[AMORO-4126][ams] Default waitAppCompletion to false on Kubernetes
Co-authored-by: caosurong <su...@thinkingdata.cn>
(cherry picked from commit eee2e3573180ae943adfa867018bdf310027efdf)
---
.../server/manager/SparkOptimizerContainer.java | 8 +++
.../manager/TestSparkOptimizerContainer.java | 73 ++++++++++++++++++++++
2 files changed, 81 insertions(+)
diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/manager/SparkOptimizerContainer.java b/amoro-ams/src/main/java/org/apache/amoro/server/manager/SparkOptimizerContainer.java
index 8609cf352..d453798fe 100644
--- a/amoro-ams/src/main/java/org/apache/amoro/server/manager/SparkOptimizerContainer.java
+++ b/amoro-ams/src/main/java/org/apache/amoro/server/manager/SparkOptimizerContainer.java
@@ -202,6 +202,12 @@ public class SparkOptimizerContainer extends AbstractOptimizerContainer {
private void addKubernetesProperties(
Resource resource, SparkOptimizerContainer.SparkConf sparkConf) {
+ sparkConf.putToOptions(
+ SparkConfKeys.KUBERNETES_SUBMISSION_WAIT_APP_COMPLETION,
+ StringUtils.defaultIfEmpty(
+ sparkConf.configValue(SparkConfKeys.KUBERNETES_SUBMISSION_WAIT_APP_COMPLETION),
+ "false"));
+
String driverName = kubernetesDriverName(resource);
sparkConf.putToOptions(
SparkOptimizerContainer.SparkConfKeys.KUBERNETES_DRIVER_NAME, driverName);
@@ -332,6 +338,8 @@ public class SparkOptimizerContainer extends AbstractOptimizerContainer {
public static final String KUBERNETES_IMAGE_REF = "spark.kubernetes.container.image";
public static final String KUBERNETES_DRIVER_NAME = "spark.kubernetes.driver.pod.name";
public static final String KUBERNETES_NAMESPACE = "spark.kubernetes.namespace";
+ public static final String KUBERNETES_SUBMISSION_WAIT_APP_COMPLETION =
+ "spark.kubernetes.submission.waitAppCompletion";
public static final String KUBERNETES_EXECUTOR_LABEL_PREFIX = "spark.kubernetes.driver.label.";
public static final String KUBERNETES_DRIVER_LABEL_PREFIX = "spark.kubernetes.driver.label.";
public static final String KUBERNETES_DRA_ENABLED = "spark.dynamicAllocation.enabled";
diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/manager/TestSparkOptimizerContainer.java b/amoro-ams/src/test/java/org/apache/amoro/server/manager/TestSparkOptimizerContainer.java
index 2936c83eb..5a432feb8 100644
--- a/amoro-ams/src/test/java/org/apache/amoro/server/manager/TestSparkOptimizerContainer.java
+++ b/amoro-ams/src/test/java/org/apache/amoro/server/manager/TestSparkOptimizerContainer.java
@@ -19,6 +19,8 @@
package org.apache.amoro.server.manager;
import org.apache.amoro.OptimizerProperties;
+import org.apache.amoro.resource.Resource;
+import org.apache.amoro.resource.ResourceType;
import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
import org.junit.Assert;
import org.junit.Test;
@@ -58,4 +60,75 @@ public class TestSparkOptimizerContainer {
Assert.assertTrue(sparkOptions.contains("--conf key2=value4"));
Assert.assertTrue(sparkOptions.contains("--conf key5=value5"));
}
+
+ @Test
+ public void testKubernetesWaitAppCompletionDefaultsToFalse() {
+ SparkOptimizerContainer container = createContainer("k8s://https://127.0.0.1:6443");
+
+ String startupArgs =
+ container.buildOptimizerStartupArgsString(createResource(Maps.newHashMap()));
+
+ Assert.assertTrue(
+ startupArgs.contains(
+ "--conf "
+ + SparkOptimizerContainer.SparkConfKeys.KUBERNETES_SUBMISSION_WAIT_APP_COMPLETION
+ + "=false"));
+ }
+
+ @Test
+ public void testKubernetesWaitAppCompletionCanBeOverridden() {
+ SparkOptimizerContainer container = createContainer("k8s://https://127.0.0.1:6443");
+ Map<String, String> resourceProperties = Maps.newHashMap();
+ resourceProperties.put(
+ SparkOptimizerContainer.SparkConf.SPARK_PARAMETER_PREFIX
+ + SparkOptimizerContainer.SparkConfKeys.KUBERNETES_SUBMISSION_WAIT_APP_COMPLETION,
+ "true");
+
+ String startupArgs =
+ container.buildOptimizerStartupArgsString(createResource(resourceProperties));
+
+ Assert.assertTrue(
+ startupArgs.contains(
+ "--conf "
+ + SparkOptimizerContainer.SparkConfKeys.KUBERNETES_SUBMISSION_WAIT_APP_COMPLETION
+ + "=true"));
+ Assert.assertFalse(
+ startupArgs.contains(
+ "--conf "
+ + SparkOptimizerContainer.SparkConfKeys.KUBERNETES_SUBMISSION_WAIT_APP_COMPLETION
+ + "=false"));
+ }
+
+ @Test
+ public void testYarnDoesNotSetKubernetesWaitAppCompletion() {
+ SparkOptimizerContainer container = createContainer("yarn");
+
+ String startupArgs =
+ container.buildOptimizerStartupArgsString(createResource(Maps.newHashMap()));
+
+ Assert.assertFalse(
+ startupArgs.contains(
+ SparkOptimizerContainer.SparkConfKeys.KUBERNETES_SUBMISSION_WAIT_APP_COMPLETION));
+ }
+
+ private SparkOptimizerContainer createContainer(String master) {
+ Map<String, String> properties = Maps.newHashMap(containerProperties);
+ properties.put(SparkOptimizerContainer.SPARK_MASTER, master);
+ if (master.startsWith("k8s://")) {
+ properties.put(
+ SparkOptimizerContainer.SparkConf.SPARK_PARAMETER_PREFIX
+ + SparkOptimizerContainer.SparkConfKeys.KUBERNETES_IMAGE_REF,
+ "apache/amoro-spark-optimizer:test");
+ }
+ SparkOptimizerContainer container = new SparkOptimizerContainer();
+ container.init("spark", properties);
+ return container;
+ }
+
+ private Resource createResource(Map<String, String> properties) {
+ return new Resource.Builder("spark", "test-group", ResourceType.OPTIMIZER)
+ .setThreadCount(1)
+ .setProperties(properties)
+ .build();
+ }
}
(amoro) 07/11: [Hotfix] Pin all external actions to the specific git hash (#4294)
Posted by xu...@apache.org.
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 6c4872763d060c2354f7534759f794590f14f9ed
Author: ZhouJinsong <zh...@163.com>
AuthorDate: Mon Jul 27 21:47:28 2026 +0800
[Hotfix] Pin all external actions to the specific git hash (#4294)
* Pin all external actions to the specific git hash
* Update action version
(cherry picked from commit ea4ea4cd31ecab35a3d8b7ebf39571d266cac472)
---
.github/workflows/docker-images.yml | 30 +++++++++++++++---------------
1 file changed, 15 insertions(+), 15 deletions(-)
diff --git a/.github/workflows/docker-images.yml b/.github/workflows/docker-images.yml
index 2f7c47bf3..2e4e946ce 100644
--- a/.github/workflows/docker-images.yml
+++ b/.github/workflows/docker-images.yml
@@ -51,13 +51,13 @@ jobs:
cache: maven
check-latest: false
- name: Set up QEMU
- uses: docker/setup-qemu-action@v2
+ uses: docker/setup-qemu-action@c7c53464625b32c7a7e944ae62b3e17d2b600130 # v3.7.0
- name: Set up Docker Buildx
- uses: docker/setup-buildx-action@v2
+ uses: docker/setup-buildx-action@f7ce87c1d6bead3e36075b2ce75da1f6cc28aaca # v3.9.0
- name: Set up Docker tags
- uses: docker/metadata-action@v5
+ uses: docker/metadata-action@318604b99e75e41977312d83839a89be02ca4893 # v5.9.0
id: meta
with:
flavor: |
@@ -76,7 +76,7 @@ jobs:
run: echo '${{ steps.meta.outputs.tags }}'
- name: Login to Docker Hub
- uses: docker/login-action@v2
+ uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # v3.7.0
with:
username: ${{ secrets.DOCKERHUB_USER }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
@@ -90,7 +90,7 @@ jobs:
run: ./mvnw clean package -pl 'dist' -am -e ${MVN_HADOOP} -DskipTests -B -ntp -Psupport-all-formats -Pno-extended-disk-storage -Pno-plugin-bin
- name: Build and Push Amoro Docker Image
- uses: docker/build-push-action@v4
+ uses: docker/build-push-action@4f58ea79222b3b9dc2c8bbdd6debcef730109a75 # v6.9.0
with:
context: .
push: true
@@ -115,13 +115,13 @@ jobs:
cache: maven
check-latest: false
- name: Set up QEMU
- uses: docker/setup-qemu-action@v2
+ uses: docker/setup-qemu-action@c7c53464625b32c7a7e944ae62b3e17d2b600130 # v3.7.0
- name: Set up Docker Buildx
- uses: docker/setup-buildx-action@v2
+ uses: docker/setup-buildx-action@f7ce87c1d6bead3e36075b2ce75da1f6cc28aaca # v3.9.0
- name: Set up Docker tags
- uses: docker/metadata-action@v5
+ uses: docker/metadata-action@318604b99e75e41977312d83839a89be02ca4893 # v5.9.0
id: meta
with:
flavor: |
@@ -141,7 +141,7 @@ jobs:
run: echo '${{ steps.meta.outputs.tags }}'
- name: Login to Docker Hub
- uses: docker/login-action@v2
+ uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # v3.7.0
with:
username: ${{ secrets.DOCKERHUB_USER }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
@@ -167,7 +167,7 @@ jobs:
run: ./mvnw clean package -pl 'amoro-optimizer/amoro-optimizer-flink' -am -e ${OPTIMIZER_FLINK} -DskipTests -B -ntp
- name: Build and Push Flink Optimizer Docker Image
- uses: docker/build-push-action@v4
+ uses: docker/build-push-action@4f58ea79222b3b9dc2c8bbdd6debcef730109a75 # v6.9.0
env:
AMORO_VERSION: ${{ steps.version.outputs.AMORO_VERSION }}
with:
@@ -198,13 +198,13 @@ jobs:
cache: maven
check-latest: false
- name: Set up QEMU
- uses: docker/setup-qemu-action@v2
+ uses: docker/setup-qemu-action@c7c53464625b32c7a7e944ae62b3e17d2b600130 # v3.7.0
- name: Set up Docker Buildx
- uses: docker/setup-buildx-action@v2
+ uses: docker/setup-buildx-action@f7ce87c1d6bead3e36075b2ce75da1f6cc28aaca # v3.9.0
- name: Set up Docker tags
- uses: docker/metadata-action@v5
+ uses: docker/metadata-action@318604b99e75e41977312d83839a89be02ca4893 # v5.9.0
id: meta
with:
flavor: |
@@ -222,7 +222,7 @@ jobs:
run: echo '${{ steps.meta.outputs.tags }}'
- name: Login to Docker Hub
- uses: docker/login-action@v2
+ uses: docker/login-action@c94ce9fb468520275223c153574b00df6fe4bcc9 # v3.7.0
with:
username: ${{ secrets.DOCKERHUB_USER }}
password: ${{ secrets.DOCKERHUB_TOKEN }}
@@ -248,7 +248,7 @@ jobs:
run: ./mvnw clean package -pl 'amoro-optimizer/amoro-optimizer-spark' -am -e ${SPARK_VERSION} -DskipTests -B -ntp
- name: Build and Push Spark Optimizer Docker Image
- uses: docker/build-push-action@v4
+ uses: docker/build-push-action@4f58ea79222b3b9dc2c8bbdd6debcef730109a75 # v6.9.0
env:
AMORO_VERSION: ${{ steps.version.outputs.AMORO_VERSION }}
SPARK_MAJOR_VERSION: ${{ steps.versions.outputs.SPARK_MAJOR_VERSION }}