This is an automated email from the ASF dual-hosted git repository. czy006 pushed a commit to branch jira/JSPT-2568 in repository https://gitbox.apache.org/repos/asf/amoro.git
commit ff305276154842813d7f2fe54a54afda160a6aed Author: ConradJam <[email protected]> AuthorDate: Thu Jul 2 14:24:58 2026 +0800 [hotfix][AMS] Paimon过期快照统一实现与Process异常恢复优化 --- .../amoro/server/process/ProcessService.java | 31 ++- .../process/executor/TableProcessExecutor.java | 28 ++- .../paimon/PaimonMaintainProcessFactory.java | 71 ++++++- .../process/TestProcessServiceRetryScheduling.java | 229 +++++++++++++++++++++ .../process/executor/TestTableProcessExecutor.java | 111 +++++++++- .../paimon/TestPaimonExpireSnapshotProcess.java | 33 +++ .../paimon/TestPaimonMaintainProcessFactory.java | 99 +++++++++ .../main/java/org/apache/amoro/PaimonActions.java | 2 +- .../process/HttpRemoteSparkStandAloneSubmit.java | 9 +- .../amoro-bin/conf/plugins/execute-engines.yaml | 7 + .../amoro-bin/conf/plugins/process-factories.yaml | 11 +- prev-conf/plugins/execute-engines.yaml | 5 +- prev-conf/plugins/process-factories.yaml | 8 +- 13 files changed, 617 insertions(+), 27 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 ac20f4e35..18dbbc72b 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 @@ -162,16 +162,27 @@ public class ProcessService extends PersistentBase { ActionCoordinatorScheduler scheduler = actionCoordinators.get(processMeta.getProcessType()); if (tableRuntime != null && scheduler != null) { - DefaultTableProcessStore store = - new DefaultTableProcessStore( - processMeta.getProcessId(), - tableRuntime, - processMeta, - scheduler.getAction(), - ActionCoordinatorScheduler.PROCESS_MAX_RETRY_NUMBER); - TableProcess process = scheduler.recover(tableRuntime, store); - trackTableProcess(tableRuntime.getTableIdentifier(), store, process); - executeOrTraceProcess(store, process); + try { + DefaultTableProcessStore store = + new DefaultTableProcessStore( + processMeta.getProcessId(), + tableRuntime, + processMeta, + scheduler.getAction(), + ActionCoordinatorScheduler.PROCESS_MAX_RETRY_NUMBER); + TableProcess process = scheduler.recover(tableRuntime, store); + trackTableProcess(tableRuntime.getTableIdentifier(), store, process); + executeOrTraceProcess(store, process); + } catch (Exception e) { + LOG.warn( + "Failed to recover active table process {}, tableId={}, processType={}," + + " executionEngine={}, skip this process recovery.", + processMeta.getProcessId(), + processMeta.getTableId(), + processMeta.getProcessType(), + processMeta.getExecutionEngine(), + e); + } } }); } 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 9bdba0a80..78e48c866 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 @@ -25,6 +25,7 @@ import org.apache.amoro.process.ProcessStatusInfo; import org.apache.amoro.process.TableProcess; import org.apache.amoro.process.TableProcessStore; import org.apache.amoro.server.persistence.PersistentBase; +import org.apache.amoro.shade.guava32.com.google.common.annotations.VisibleForTesting; import org.apache.amoro.shade.guava32.com.google.common.base.Strings; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -43,6 +44,7 @@ public class TableProcessExecutor extends PersistentBase implements Runnable { public ExecuteEngine executeEngine; protected TableProcess tableProcess; private final TableProcessStore store; + private final long pollIntervalMs; private Runnable finishedCallback; /** @@ -53,9 +55,19 @@ public class TableProcessExecutor extends PersistentBase implements Runnable { */ public TableProcessExecutor( TableProcess tableProcess, TableProcessStore store, ExecuteEngine executeEngine) { + this(tableProcess, store, executeEngine, DEFAULT_POLL_INTERVAL_MS); + } + + @VisibleForTesting + TableProcessExecutor( + TableProcess tableProcess, + TableProcessStore store, + ExecuteEngine executeEngine, + long pollIntervalMs) { this.tableProcess = tableProcess; this.executeEngine = executeEngine; this.store = store; + this.pollIntervalMs = pollIntervalMs; } /** Submit or recover the process to engine, poll status and update store. */ @@ -113,12 +125,13 @@ public class TableProcessExecutor extends PersistentBase implements Runnable { return; } try { - Thread.sleep(DEFAULT_POLL_INTERVAL_MS); + Thread.sleep(pollIntervalMs); } catch (InterruptedException e) { throw e; } statusInfo = executeEngine.getStatusInfo(externalProcessIdentifier); status = statusInfo.getStatus(); + persistRunningStatusIfNecessary(status, externalProcessIdentifier); } message = statusInfo.getMessage(); } catch (Throwable t) { @@ -237,6 +250,19 @@ public class TableProcessExecutor extends PersistentBase implements Runnable { || status == ProcessStatus.SUBMITTED); } + private void persistRunningStatusIfNecessary( + ProcessStatus status, String externalProcessIdentifier) { + if (status == ProcessStatus.RUNNING && store.getStatus() != ProcessStatus.RUNNING) { + store.tryTransitState( + ProcessStatus.RUNNING, + ProcessEvent.SUBMIT_REQUESTED, + externalProcessIdentifier, + "Process is running.", + tableProcess.getProcessParameters(), + tableProcess.getSummary()); + } + } + static String buildFailureMessage(Throwable throwable) { if (throwable == null) { return ""; diff --git a/amoro-ams/src/main/java/org/apache/amoro/server/process/paimon/PaimonMaintainProcessFactory.java b/amoro-ams/src/main/java/org/apache/amoro/server/process/paimon/PaimonMaintainProcessFactory.java index a8613d86b..52087a8fe 100644 --- a/amoro-ams/src/main/java/org/apache/amoro/server/process/paimon/PaimonMaintainProcessFactory.java +++ b/amoro-ams/src/main/java/org/apache/amoro/server/process/paimon/PaimonMaintainProcessFactory.java @@ -26,6 +26,7 @@ import org.apache.amoro.config.ConfigOption; import org.apache.amoro.config.ConfigOptions; import org.apache.amoro.config.Configurations; import org.apache.amoro.process.ExecuteEngine; +import org.apache.amoro.process.HttpRemoteSparkStandAloneSubmit; import org.apache.amoro.process.LocalExecutionEngine; import org.apache.amoro.process.ProcessFactory; import org.apache.amoro.process.ProcessTriggerStrategy; @@ -42,11 +43,11 @@ import java.util.Collections; import java.util.Map; import java.util.Optional; import java.util.Set; +import java.util.stream.Collectors; /** - * Process factory for Paimon table <em>maintenance</em> actions (currently metadata synchronization - * only; the snapshot-expiration path lives on the {@code czy006/paimon-factory-refactor} feature - * branch and is intentionally excluded from {@code dev-paimon-compact} until that branch merges). + * Process factory for Paimon table <em>maintenance</em> actions, including metadata synchronization + * and remote snapshot expiration. * * <p>Renamed from {@code PaimonProcessFactory} in AMORO-4200 to disambiguate from the optimizing * factory {@code org.apache.amoro.formats.paimon.process.PaimonProcessFactory} in the {@code @@ -71,11 +72,34 @@ public class PaimonMaintainProcessFactory implements ProcessFactory { public static final ConfigOption<Integer> SYNC_TABLE_META_TRIGGER_PARALLELISM = ConfigOptions.key("sync-table-meta.trigger-parallelism").intType().defaultValue(1); + public static final ConfigOption<Boolean> SNAPSHOT_EXPIRE_ENABLED = + ConfigOptions.key("expire-snapshots.enabled").booleanType().defaultValue(false); + + public static final ConfigOption<Duration> SNAPSHOT_EXPIRE_INTERVAL = + ConfigOptions.key("expire-snapshots.interval") + .durationType() + .defaultValue(Duration.ofHours(24)); + + public static final ConfigOption<Integer> SPARK_VERSION = + ConfigOptions.key("spark-version").intType().defaultValue(354); + private final Map<Action, ProcessTriggerStrategy> actions = Maps.newHashMap(); + private ExecuteEngine remoteEngine; + private int sparkVersion = SPARK_VERSION.defaultValue(); @Override public void availableExecuteEngines(Collection<ExecuteEngine> allAvailableEngines) { - // No remote engines needed — sync-meta is an in-process periodic job. + this.remoteEngine = null; + if (allAvailableEngines == null) { + return; + } + for (ExecuteEngine engine : allAvailableEngines) { + if (engine instanceof HttpRemoteSparkStandAloneSubmit + || HttpRemoteSparkStandAloneSubmit.ENGINE_NAME.equals(engine.name())) { + this.remoteEngine = engine; + return; + } + } } @Override @@ -98,6 +122,10 @@ public class PaimonMaintainProcessFactory implements ProcessFactory { return Optional.of(new PaimonTableMetaSyncProcess(tableRuntime)); } + if (PaimonActions.EXPIRE_SNAPSHOTS.equals(action)) { + return triggerExpireSnapshots(tableRuntime); + } + return Optional.empty(); } @@ -107,6 +135,15 @@ public class PaimonMaintainProcessFactory implements ProcessFactory { if (PaimonActions.SYNC_TABLE_META.equals(store.getAction())) { return new PaimonTableMetaSyncProcess(tableRuntime); } + if (PaimonActions.EXPIRE_SNAPSHOTS.equals(store.getAction())) { + if (remoteEngine == null) { + throw new RecoverProcessFailedException( + "Cannot recover Paimon expire snapshots process without " + + HttpRemoteSparkStandAloneSubmit.ENGINE_NAME + + " execute engine"); + } + return new PaimonExpireSnapshotProcess(tableRuntime, remoteEngine, sparkVersion); + } throw new RecoverProcessFailedException( "Unsupported action for PaimonMaintainProcessFactory: " + store.getAction()); } @@ -124,7 +161,15 @@ public class PaimonMaintainProcessFactory implements ProcessFactory { PaimonActions.SYNC_TABLE_META, new ProcessTriggerStrategy(interval, false, Math.max(parallelism, 1))); } - LOG.info("Apache Paimon Process Factory initialized actions {}.", this.actions); + if (configs.getBoolean(SNAPSHOT_EXPIRE_ENABLED)) { + Duration interval = configs.getDuration(SNAPSHOT_EXPIRE_INTERVAL); + this.sparkVersion = configs.getInteger(SPARK_VERSION); + this.actions.put( + PaimonActions.EXPIRE_SNAPSHOTS, ProcessTriggerStrategy.triggerAtFixRate(interval)); + } + LOG.info( + "Apache Paimon Process Factory initialized actions {}.", + this.actions.keySet().stream().map(Action::getName).collect(Collectors.toList())); } @Override @@ -143,5 +188,21 @@ public class PaimonMaintainProcessFactory implements ProcessFactory { private void resetConfiguredState() { actions.clear(); + remoteEngine = null; + sparkVersion = SPARK_VERSION.defaultValue(); + } + + private Optional<TableProcess> triggerExpireSnapshots(TableRuntime tableRuntime) { + if (remoteEngine == null) { + LOG.warn( + "Skip Paimon expire snapshots for table {} because execute engine {} is not installed.", + tableRuntime.getTableIdentifier(), + HttpRemoteSparkStandAloneSubmit.ENGINE_NAME); + return Optional.empty(); + } + ProcessTriggerStrategy strategy = actions.get(PaimonActions.EXPIRE_SNAPSHOTS); + return PaimonExpireSnapshotProcess.trigger( + tableRuntime, remoteEngine, sparkVersion, strategy.getTriggerInterval()) + .map(process -> (TableProcess) process); } } diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/process/TestProcessServiceRetryScheduling.java b/amoro-ams/src/test/java/org/apache/amoro/server/process/TestProcessServiceRetryScheduling.java index b003d3c85..e2a8822f8 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/process/TestProcessServiceRetryScheduling.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/process/TestProcessServiceRetryScheduling.java @@ -25,6 +25,7 @@ import org.apache.amoro.TableFormat; import org.apache.amoro.TableRuntime; import org.apache.amoro.config.TableConfiguration; import org.apache.amoro.metrics.MetricRegistry; +import org.apache.amoro.process.ActionCoordinator; import org.apache.amoro.process.EngineType; import org.apache.amoro.process.ExecuteEngine; import org.apache.amoro.process.ProcessStatus; @@ -36,9 +37,11 @@ import org.junit.Assert; import org.junit.Test; import java.lang.reflect.Field; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; @@ -46,10 +49,13 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; +import java.util.function.BooleanSupplier; import java.util.function.Function; public class TestProcessServiceRetryScheduling extends AMSManagerTestBase { private static final long RETRY_ASSERT_TIMEOUT_SECONDS = 8L; + private static final Action RECOVER_FAIL_ACTION = Action.register("recover_fail"); + private static final Action RECOVER_OK_ACTION = Action.register("recover_ok"); @Test(timeout = 20_000) public void testRetryDelayShouldNotBlockProcessExecutionThreads() throws Exception { @@ -80,6 +86,119 @@ public class TestProcessServiceRetryScheduling extends AMSManagerTestBase { } } + @Test(timeout = 20_000) + public void testRecoverProcessesShouldSkipFailedRecovery() throws Exception { + ProcessService processService = new ProcessService(null); + AlwaysRunningExecuteEngine executeEngine = new AlwaysRunningExecuteEngine(); + processService.installExecuteEngine(executeEngine); + + try { + TableRuntime runtime = new TestingTableRuntime(20_001L); + processService.register( + runtime, new TestingTableProcess(runtime, executeEngine, RECOVER_FAIL_ACTION)); + ProcessService.TableProcessHolder holder = awaitSingleActiveProcess(processService); + processService.untrackTableProcessInstance( + runtime.getTableIdentifier(), holder.getStore().getProcessId()); + + processService.installActionCoordinator(new ThrowingRecoverCoordinator(RECOVER_FAIL_ACTION)); + + processService.recoverProcesses(Collections.singletonList(runtime)); + + Assert.assertTrue(processService.getActiveTableProcess().isEmpty()); + } finally { + shutdownExecutor(processService, "processExecutionPool"); + shutdownExecutor(processService, "retrySchedulingPool"); + processService.dispose(); + } + } + + @Test(timeout = 20_000) + public void testRecoverProcessesContinuesAfterSingleFailure() throws Exception { + ProcessService processService = new ProcessService(null); + AlwaysRunningExecuteEngine executeEngine = new AlwaysRunningExecuteEngine(); + processService.installExecuteEngine(executeEngine); + + try { + TableRuntime failedRuntime = new TestingTableRuntime(20_002L); + TableRuntime recoveredRuntime = new TestingTableRuntime(20_003L); + processService.register( + failedRuntime, + new TestingTableProcess(failedRuntime, executeEngine, RECOVER_FAIL_ACTION)); + ProcessService.TableProcessHolder failedHolder = awaitSingleActiveProcess(processService); + processService.register( + recoveredRuntime, + new TestingTableProcess(recoveredRuntime, executeEngine, RECOVER_OK_ACTION)); + awaitActiveProcessCount(processService, 2); + + processService.untrackTableProcessInstance( + failedRuntime.getTableIdentifier(), failedHolder.getStore().getProcessId()); + ProcessService.TableProcessHolder recoveredHolder = + processService + .getTableProcessInstances(recoveredRuntime.getTableIdentifier()) + .values() + .iterator() + .next(); + processService.untrackTableProcessInstance( + recoveredRuntime.getTableIdentifier(), recoveredHolder.getStore().getProcessId()); + + processService.installActionCoordinator(new ThrowingRecoverCoordinator(RECOVER_FAIL_ACTION)); + processService.installActionCoordinator( + new RecoveringCoordinator(RECOVER_OK_ACTION, executeEngine)); + + processService.recoverProcesses(Arrays.asList(failedRuntime, recoveredRuntime)); + + Assert.assertTrue( + processService.getTableProcessInstances(failedRuntime.getTableIdentifier()).isEmpty()); + Assert.assertEquals( + 1, processService.getTableProcessInstances(recoveredRuntime.getTableIdentifier()).size()); + } finally { + shutdownExecutor(processService, "processExecutionPool"); + shutdownExecutor(processService, "retrySchedulingPool"); + processService.dispose(); + } + } + + private ProcessService.TableProcessHolder awaitSingleActiveProcess(ProcessService processService) + throws InterruptedException { + awaitActiveProcessCount(processService, 1); + return processService.getActiveTableProcess().values().stream() + .flatMap(processes -> processes.values().stream()) + .findFirst() + .orElseThrow(() -> new AssertionError("No active process found")); + } + + private void awaitActiveProcessCount(ProcessService processService, int expectedCount) + throws InterruptedException { + awaitCondition( + () -> { + List<ProcessService.TableProcessHolder> activeHolders = + processService.getActiveTableProcess().values().stream() + .flatMap(processes -> processes.values().stream()) + .collect(java.util.stream.Collectors.toList()); + long recoverableProcessCount = + activeHolders.stream() + .filter( + holder -> + holder.getStore().getStatus() == ProcessStatus.SUBMITTED + || holder.getStore().getStatus() == ProcessStatus.RUNNING) + .count(); + return activeHolders.size() >= expectedCount && recoverableProcessCount >= expectedCount; + }, + RETRY_ASSERT_TIMEOUT_SECONDS * 1000L, + 100L); + } + + private void awaitCondition(BooleanSupplier condition, long maxWaitMs, long intervalMs) + throws InterruptedException { + long start = System.currentTimeMillis(); + while (!condition.getAsBoolean()) { + if (System.currentTimeMillis() - start >= maxWaitMs) { + throw new AssertionError("Condition not met within " + maxWaitMs + " ms"); + } + Thread.sleep(intervalMs); + } + } + private void shutdownExecutor(ProcessService processService, String fieldName) { try { Field field = ProcessService.class.getDeclaredField(fieldName); @@ -148,6 +267,116 @@ public class TestProcessServiceRetryScheduling extends AMSManagerTestBase { } } + private static class AlwaysRunningExecuteEngine implements ExecuteEngine { + private final AtomicLong identifierGenerator = new AtomicLong(0L); + + @Override + public EngineType engineType() { + return EngineType.of(name()); + } + + @Override + public ProcessStatus getStatus(String processIdentifier) { + return ProcessStatus.RUNNING; + } + + @Override + public String submitTableProcess(TableProcess tableProcess) { + return "running-process-" + identifierGenerator.incrementAndGet(); + } + + @Override + public ProcessStatus tryCancelTableProcess( + TableProcess tableProcess, String processIdentifier) { + return ProcessStatus.CANCELING; + } + + @Override + public void open(Map<String, String> properties) {} + + @Override + public void close() {} + + @Override + public String name() { + return "recover-running"; + } + } + + private static class RecoveringCoordinator implements ActionCoordinator { + private final Action action; + private final ExecuteEngine executeEngine; + + private RecoveringCoordinator(Action action, ExecuteEngine executeEngine) { + this.action = action; + this.executeEngine = executeEngine; + } + + @Override + public boolean formatSupported(TableFormat format) { + return true; + } + + @Override + public int parallelism() { + return 1; + } + + @Override + public Action action() { + return action; + } + + @Override + public long getNextExecutingTime(TableRuntime tableRuntime) { + return 24 * 60 * 60 * 1000L; + } + + @Override + public boolean enabled(TableRuntime tableRuntime) { + return true; + } + + @Override + public long getExecutorDelay() { + return 0L; + } + + @Override + public Optional<TableProcess> trigger(TableRuntime tableRuntime) { + return Optional.empty(); + } + + @Override + public TableProcess recoverTableProcess( + TableRuntime tableRuntime, TableProcessStore processStore) { + return new TestingTableProcess(tableRuntime, executeEngine, action); + } + + @Override + public void open(Map<String, String> properties) {} + + @Override + public void close() {} + + @Override + public String name() { + return "recovering-" + action.getName(); + } + } + + private static class ThrowingRecoverCoordinator extends RecoveringCoordinator { + private ThrowingRecoverCoordinator(Action action) { + super(action, new AlwaysRunningExecuteEngine()); + } + + @Override + public TableProcess recoverTableProcess( + TableRuntime tableRuntime, TableProcessStore processStore) { + throw new IllegalStateException("recover failed for test"); + } + } + private static class TestingTableProcess extends TableProcess { private final Action action; diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/process/executor/TestTableProcessExecutor.java b/amoro-ams/src/test/java/org/apache/amoro/server/process/executor/TestTableProcessExecutor.java index 13c0a02c0..7c75c854f 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/process/executor/TestTableProcessExecutor.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/process/executor/TestTableProcessExecutor.java @@ -167,16 +167,97 @@ public class TestTableProcessExecutor { Assert.assertEquals("qid-1", store.getExternalProcessIdentifier()); Assert.assertTrue(store.getFinishTime() > 0); Assert.assertEquals( - Arrays.asList(ProcessEvent.SUBMIT_REQUESTED.name(), ProcessEvent.COMPLETE_SUCCESS.name()), + Arrays.asList( + ProcessEvent.SUBMIT_REQUESTED.name(), + ProcessEvent.SUBMIT_REQUESTED.name(), + ProcessEvent.COMPLETE_SUCCESS.name()), store.getEvents()); Assert.assertEquals(3, engine.getPollCount()); } + @Test + public void testPersistRunningStatusAfterSubmittedPoll() { + InMemoryTableProcessStore store = new InMemoryTableProcessStore(); + SequencedExecuteEngine engine = + new SequencedExecuteEngine( + ProcessStatusInfo.of(ProcessStatus.SUBMITTED, ""), + Arrays.asList( + ProcessStatusInfo.of(ProcessStatus.RUNNING, ""), + ProcessStatusInfo.of(ProcessStatus.SUCCESS, ""))); + + TableProcessExecutor executor = + new TableProcessExecutor(new TestingTableProcess(), store, engine, 10L); + + executor.run(); + + Assert.assertEquals(ProcessStatus.SUCCESS, store.getStatus()); + Assert.assertEquals("qid-1", store.getExternalProcessIdentifier()); + Assert.assertEquals(1, engine.getSubmitCount()); + Assert.assertEquals( + Collections.singletonList(ProcessStatus.RUNNING), + store.transitionsTo(ProcessStatus.RUNNING)); + Assert.assertEquals( + Arrays.asList(ProcessStatus.SUBMITTED, ProcessStatus.RUNNING, ProcessStatus.SUCCESS), + store.getTransitions()); + } + + @Test + public void testRecoveredSubmittedProcessPersistsRunningWithoutResubmit() { + InMemoryTableProcessStore store = + new InMemoryTableProcessStore(ProcessStatus.SUBMITTED, "qid-1"); + SequencedExecuteEngine engine = + new SequencedExecuteEngine( + ProcessStatusInfo.of(ProcessStatus.SUBMITTED, ""), + Arrays.asList( + ProcessStatusInfo.of(ProcessStatus.RUNNING, ""), + ProcessStatusInfo.of(ProcessStatus.SUCCESS, ""))); + + TableProcessExecutor executor = + new TableProcessExecutor(new TestingTableProcess(), store, engine, 10L); + + executor.run(); + + Assert.assertEquals(ProcessStatus.SUCCESS, store.getStatus()); + Assert.assertEquals("qid-1", store.getExternalProcessIdentifier()); + Assert.assertEquals(0, engine.getSubmitCount()); + Assert.assertEquals( + Collections.singletonList(ProcessStatus.RUNNING), + store.transitionsTo(ProcessStatus.RUNNING)); + Assert.assertEquals( + Arrays.asList(ProcessStatus.RUNNING, ProcessStatus.SUCCESS), store.getTransitions()); + } + + @Test + public void testPersistRunningStatusOnlyOnce() { + InMemoryTableProcessStore store = new InMemoryTableProcessStore(); + SequencedExecuteEngine engine = + new SequencedExecuteEngine( + ProcessStatusInfo.of(ProcessStatus.SUBMITTED, ""), + Arrays.asList( + ProcessStatusInfo.of(ProcessStatus.RUNNING, ""), + ProcessStatusInfo.of(ProcessStatus.RUNNING, ""), + ProcessStatusInfo.of(ProcessStatus.SUCCESS, ""))); + + TableProcessExecutor executor = + new TableProcessExecutor(new TestingTableProcess(), store, engine, 10L); + + executor.run(); + + Assert.assertEquals(ProcessStatus.SUCCESS, store.getStatus()); + Assert.assertEquals( + Collections.singletonList(ProcessStatus.RUNNING), + store.transitionsTo(ProcessStatus.RUNNING)); + Assert.assertEquals( + Arrays.asList(ProcessStatus.SUBMITTED, ProcessStatus.RUNNING, ProcessStatus.SUCCESS), + store.getTransitions()); + } + private static class SequencedExecuteEngine implements ExecuteEngine { private final ProcessStatusInfo firstStatus; private final List<ProcessStatusInfo> remainingStatuses; private final AtomicInteger pollCount = new AtomicInteger(); + private final AtomicInteger submitCount = new AtomicInteger(); private final AtomicInteger callSequence; private int firstPollOrder = -1; @@ -222,6 +303,7 @@ public class TestTableProcessExecutor { @Override public String submitTableProcess(TableProcess tableProcess) { + submitCount.incrementAndGet(); return "qid-1"; } @@ -246,6 +328,10 @@ public class TestTableProcessExecutor { return pollCount.get(); } + public int getSubmitCount() { + return submitCount.get(); + } + public int getFirstPollOrder() { return firstPollOrder; } @@ -316,6 +402,14 @@ public class TestTableProcessExecutor { private String failMessage = ""; private long finishTime = 0L; private final List<String> events = new ArrayList<>(); + private final List<ProcessStatus> transitions = new ArrayList<>(); + + private InMemoryTableProcessStore() {} + + private InMemoryTableProcessStore(ProcessStatus status, String externalProcessIdentifier) { + this.status = status; + this.externalProcessIdentifier = externalProcessIdentifier; + } @Override public long getProcessId() { @@ -405,6 +499,7 @@ public class TestTableProcessExecutor { return false; } events.add(processEvent.name()); + transitions.add(newStatus); this.status = newStatus; this.externalProcessIdentifier = externalProcessIdentifier; if (processEvent == ProcessEvent.COMPLETE_FAILED) { @@ -470,6 +565,20 @@ public class TestTableProcessExecutor { public List<String> getEvents() { return events; } + + public List<ProcessStatus> getTransitions() { + return transitions; + } + + public List<ProcessStatus> transitionsTo(ProcessStatus expected) { + List<ProcessStatus> result = new ArrayList<>(); + for (ProcessStatus transition : transitions) { + if (transition == expected) { + result.add(transition); + } + } + return result; + } } private static class OrderedInMemoryTableProcessStore extends InMemoryTableProcessStore { diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/process/paimon/TestPaimonExpireSnapshotProcess.java b/amoro-ams/src/test/java/org/apache/amoro/server/process/paimon/TestPaimonExpireSnapshotProcess.java index e021f45f8..74c88d3e3 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/process/paimon/TestPaimonExpireSnapshotProcess.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/process/paimon/TestPaimonExpireSnapshotProcess.java @@ -22,6 +22,9 @@ import org.apache.amoro.ServerTableIdentifier; import org.apache.amoro.TableFormat; import org.apache.amoro.TableRuntime; import org.apache.amoro.process.HttpRemoteSparkStandAloneSubmit; +import org.apache.amoro.process.ProcessStatus; +import org.apache.amoro.server.table.DefaultTableRuntime; +import org.apache.amoro.server.table.cleanup.CleanupOperation; import org.junit.Assert; import org.junit.Test; import org.mockito.Mockito; @@ -55,6 +58,20 @@ public class TestPaimonExpireSnapshotProcess { "{\"sparkVersion\":\"354\",\"paimon.version\":\"1.3\"}", params.get("conf")); } + @Test + public void testActionNameUsesPluralExpireSnapshots() { + TableRuntime runtime = Mockito.mock(TableRuntime.class); + Mockito.when(runtime.getTableIdentifier()) + .thenReturn(ServerTableIdentifier.of("catalog", "db", "tbl", TableFormat.PAIMON)); + + HttpRemoteSparkStandAloneSubmit engine = new HttpRemoteSparkStandAloneSubmit(); + engine.open(Collections.singletonMap("execute.user", "amoro")); + + PaimonExpireSnapshotProcess process = new PaimonExpireSnapshotProcess(runtime, engine, 354); + + Assert.assertEquals("EXPIRE-SNAPSHOTS", process.getAction().getName()); + } + @Test public void testBuildExpireSnapshotsSqlContainsMaxDeletes() { TableRuntime runtime = Mockito.mock(TableRuntime.class); @@ -77,4 +94,20 @@ public class TestPaimonExpireSnapshotProcess { Assert.assertTrue(sql.contains("retain_max => 12")); Assert.assertTrue(sql.contains("max_deletes => 550")); } + + @Test + public void testAfterCompleteSuccessUpdatesLastCleanTime() { + DefaultTableRuntime runtime = Mockito.mock(DefaultTableRuntime.class); + Mockito.when(runtime.getTableIdentifier()) + .thenReturn(ServerTableIdentifier.of("catalog", "default", "orders", TableFormat.PAIMON)); + + HttpRemoteSparkStandAloneSubmit engine = new HttpRemoteSparkStandAloneSubmit(); + engine.open(Collections.singletonMap("execute.user", "amoro")); + + PaimonExpireSnapshotProcess process = new PaimonExpireSnapshotProcess(runtime, engine, 354); + process.afterComplete(ProcessStatus.SUCCESS); + + Mockito.verify(runtime) + .updateLastCleanTime(Mockito.eq(CleanupOperation.SNAPSHOTS_EXPIRING), Mockito.anyLong()); + } } diff --git a/amoro-ams/src/test/java/org/apache/amoro/server/process/paimon/TestPaimonMaintainProcessFactory.java b/amoro-ams/src/test/java/org/apache/amoro/server/process/paimon/TestPaimonMaintainProcessFactory.java index bef216318..75d99a9c9 100644 --- a/amoro-ams/src/test/java/org/apache/amoro/server/process/paimon/TestPaimonMaintainProcessFactory.java +++ b/amoro-ams/src/test/java/org/apache/amoro/server/process/paimon/TestPaimonMaintainProcessFactory.java @@ -19,15 +19,19 @@ package org.apache.amoro.server.process.paimon; import org.apache.amoro.PaimonActions; +import org.apache.amoro.ServerTableIdentifier; import org.apache.amoro.TableFormat; +import org.apache.amoro.process.HttpRemoteSparkStandAloneSubmit; import org.apache.amoro.process.LocalExecutionEngine; import org.apache.amoro.process.ProcessTriggerStrategy; import org.apache.amoro.process.TableProcess; +import org.apache.amoro.process.TableProcessStore; import org.apache.amoro.server.table.DefaultTableRuntime; import org.junit.Assert; import org.junit.Test; import org.mockito.Mockito; +import java.time.Duration; import java.util.Collections; import java.util.HashMap; import java.util.Map; @@ -55,6 +59,25 @@ public class TestPaimonMaintainProcessFactory { Assert.assertEquals(2 * 60 * 60 * 1000L, strategy.getTriggerInterval().toMillis()); } + @Test + public void testExpireSnapshotsActionAndTriggerStrategy() { + PaimonMaintainProcessFactory factory = new PaimonMaintainProcessFactory(); + Map<String, String> properties = new HashMap<>(); + properties.put("sync-table-meta.enabled", "false"); + properties.put("expire-snapshots.enabled", "true"); + properties.put("expire-snapshots.interval", "24h"); + properties.put("spark-version", "321"); + factory.open(properties); + + Assert.assertTrue( + factory.supportedActions().getOrDefault(TableFormat.PAIMON, Collections.emptySet()).stream() + .anyMatch(action -> action.equals(PaimonActions.EXPIRE_SNAPSHOTS))); + + ProcessTriggerStrategy strategy = + factory.triggerStrategy(TableFormat.PAIMON, PaimonActions.EXPIRE_SNAPSHOTS); + Assert.assertEquals(Duration.ofHours(24), strategy.getTriggerInterval()); + } + @Test public void testTriggerAndRecoverUseLocalEngine() throws Exception { PaimonMaintainProcessFactory factory = new PaimonMaintainProcessFactory(); @@ -68,6 +91,80 @@ public class TestPaimonMaintainProcessFactory { Assert.assertEquals(LocalExecutionEngine.ENGINE_NAME, process.get().getExecutionEngine()); } + @Test + public void testTriggerExpireSnapshotsUseHttpSparkEngine() { + PaimonMaintainProcessFactory factory = new PaimonMaintainProcessFactory(); + Map<String, String> properties = new HashMap<>(); + properties.put("sync-table-meta.enabled", "false"); + properties.put("expire-snapshots.enabled", "true"); + properties.put("expire-snapshots.interval", "24h"); + properties.put("spark-version", "321"); + factory.open(properties); + + HttpRemoteSparkStandAloneSubmit engine = new HttpRemoteSparkStandAloneSubmit(); + engine.open(Collections.singletonMap("execute.user", "amoro")); + factory.availableExecuteEngines(Collections.singletonList(engine)); + + DefaultTableRuntime runtime = Mockito.mock(DefaultTableRuntime.class); + Mockito.when(runtime.getFormat()).thenReturn(TableFormat.PAIMON); + Mockito.when(runtime.getTableIdentifier()) + .thenReturn(ServerTableIdentifier.of("catalog", "db", "tbl", TableFormat.PAIMON)); + Mockito.when(runtime.getTableConfig()).thenReturn(Collections.emptyMap()); + Mockito.when(runtime.getLastCleanTime(Mockito.any())).thenReturn(0L); + + Optional<TableProcess> process = factory.trigger(runtime, PaimonActions.EXPIRE_SNAPSHOTS); + + Assert.assertTrue(process.isPresent()); + Assert.assertTrue(process.get() instanceof PaimonExpireSnapshotProcess); + Assert.assertEquals( + HttpRemoteSparkStandAloneSubmit.ENGINE_NAME, process.get().getExecutionEngine()); + Assert.assertEquals("321", process.get().getProcessParameters().get("sparkVersion")); + } + + @Test + public void testTriggerExpireSnapshotsWithoutHttpSparkEngineReturnsEmpty() { + PaimonMaintainProcessFactory factory = new PaimonMaintainProcessFactory(); + Map<String, String> properties = new HashMap<>(); + properties.put("sync-table-meta.enabled", "false"); + properties.put("expire-snapshots.enabled", "true"); + factory.open(properties); + factory.availableExecuteEngines(Collections.singletonList(new LocalExecutionEngine())); + + DefaultTableRuntime runtime = Mockito.mock(DefaultTableRuntime.class); + Mockito.when(runtime.getFormat()).thenReturn(TableFormat.PAIMON); + + Optional<TableProcess> process = factory.trigger(runtime, PaimonActions.EXPIRE_SNAPSHOTS); + + Assert.assertFalse(process.isPresent()); + } + + @Test + public void testRecoverExpireSnapshotsUseHttpSparkEngine() throws Exception { + PaimonMaintainProcessFactory factory = new PaimonMaintainProcessFactory(); + Map<String, String> properties = new HashMap<>(); + properties.put("sync-table-meta.enabled", "false"); + properties.put("expire-snapshots.enabled", "true"); + properties.put("spark-version", "321"); + factory.open(properties); + + HttpRemoteSparkStandAloneSubmit engine = new HttpRemoteSparkStandAloneSubmit(); + engine.open(Collections.singletonMap("execute.user", "amoro")); + factory.availableExecuteEngines(Collections.singletonList(engine)); + + DefaultTableRuntime runtime = Mockito.mock(DefaultTableRuntime.class); + Mockito.when(runtime.getTableIdentifier()) + .thenReturn(ServerTableIdentifier.of("catalog", "db", "tbl", TableFormat.PAIMON)); + Mockito.when(runtime.getTableConfig()).thenReturn(Collections.emptyMap()); + TableProcessStore store = Mockito.mock(TableProcessStore.class); + Mockito.when(store.getAction()).thenReturn(PaimonActions.EXPIRE_SNAPSHOTS); + + TableProcess process = factory.recover(runtime, store); + + Assert.assertTrue(process instanceof PaimonExpireSnapshotProcess); + Assert.assertEquals(HttpRemoteSparkStandAloneSubmit.ENGINE_NAME, process.getExecutionEngine()); + Assert.assertEquals("321", process.getProcessParameters().get("sparkVersion")); + } + @Test public void testOpenWithEmptyPropertiesUseDefaults() { PaimonMaintainProcessFactory factory = new PaimonMaintainProcessFactory(); @@ -76,6 +173,7 @@ public class TestPaimonMaintainProcessFactory { Set<org.apache.amoro.Action> actions = factory.supportedActions().getOrDefault(TableFormat.PAIMON, Collections.emptySet()); Assert.assertTrue(actions.contains(PaimonActions.SYNC_TABLE_META)); + Assert.assertFalse(actions.contains(PaimonActions.EXPIRE_SNAPSHOTS)); ProcessTriggerStrategy syncStrategy = factory.triggerStrategy(TableFormat.PAIMON, PaimonActions.SYNC_TABLE_META); @@ -95,6 +193,7 @@ public class TestPaimonMaintainProcessFactory { Set<org.apache.amoro.Action> actions = factory.supportedActions().getOrDefault(TableFormat.PAIMON, Collections.emptySet()); Assert.assertFalse(actions.contains(PaimonActions.SYNC_TABLE_META)); + Assert.assertFalse(actions.contains(PaimonActions.EXPIRE_SNAPSHOTS)); } @Test diff --git a/amoro-common/src/main/java/org/apache/amoro/PaimonActions.java b/amoro-common/src/main/java/org/apache/amoro/PaimonActions.java index d1174a48d..a76d081c1 100644 --- a/amoro-common/src/main/java/org/apache/amoro/PaimonActions.java +++ b/amoro-common/src/main/java/org/apache/amoro/PaimonActions.java @@ -21,7 +21,7 @@ package org.apache.amoro; public class PaimonActions { public static final Action SYNC_TABLE_META = Action.register("sync-table-meta"); - public static final Action EXPIRE_SNAPSHOTS = Action.register("expire-snapshot"); + public static final Action EXPIRE_SNAPSHOTS = Action.register("expire-snapshots"); private PaimonActions() {} } diff --git a/amoro-common/src/main/java/org/apache/amoro/process/HttpRemoteSparkStandAloneSubmit.java b/amoro-common/src/main/java/org/apache/amoro/process/HttpRemoteSparkStandAloneSubmit.java index 850469336..51595e0cc 100644 --- a/amoro-common/src/main/java/org/apache/amoro/process/HttpRemoteSparkStandAloneSubmit.java +++ b/amoro-common/src/main/java/org/apache/amoro/process/HttpRemoteSparkStandAloneSubmit.java @@ -83,6 +83,9 @@ public class HttpRemoteSparkStandAloneSubmit implements ExecuteEngine { private static final String PARAM_CONF = "conf"; private static final String PARAM_CLIENT_IP = "clientIp"; + private static final String DEFAULT_SPARK_SESSION_SQL_PREFIX = + "set spark.syntax.extension=true; set spark.paimon.version=1.3;"; + private final ObjectMapper objectMapper = new ObjectMapper(); private String baseUrl; @@ -292,7 +295,7 @@ public class HttpRemoteSparkStandAloneSubmit implements ExecuteEngine { ObjectNode requestNode = objectMapper.createObjectNode(); requestNode.put(PARAM_JOB_TYPE, "sql"); - requestNode.put(PARAM_HQL, hql); + requestNode.put(PARAM_HQL, prependDefaultSparkSessionSql(hql)); requestNode.put(PARAM_CUR_USER, params.getOrDefault(PARAM_CUR_USER, executeUser)); requestNode.put(PARAM_SOURCE_TAG, params.getOrDefault(PARAM_SOURCE_TAG, sourceTag)); requestNode.put( @@ -316,6 +319,10 @@ public class HttpRemoteSparkStandAloneSubmit implements ExecuteEngine { } } + private String prependDefaultSparkSessionSql(String hql) { + return DEFAULT_SPARK_SESSION_SQL_PREFIX + "\n" + hql; + } + private String doPost(String path, String jsonBody) { HttpURLConnection connection = null; try { diff --git a/dist/src/main/amoro-bin/conf/plugins/execute-engines.yaml b/dist/src/main/amoro-bin/conf/plugins/execute-engines.yaml index 5c3199f32..5a3cca278 100755 --- a/dist/src/main/amoro-bin/conf/plugins/execute-engines.yaml +++ b/dist/src/main/amoro-bin/conf/plugins/execute-engines.yaml @@ -22,3 +22,10 @@ execute-engines: properties: pool.default.thread-count: 10 pool.snapshots-expiring.thread-count: 10 + - name: sl-spark-http + enabled: true + priority: 100 + properties: + execute.user: amoro + default-spark-version: 354 + source-tag: AMORO diff --git a/dist/src/main/amoro-bin/conf/plugins/process-factories.yaml b/dist/src/main/amoro-bin/conf/plugins/process-factories.yaml index d64a91f1d..f3bd7cf4e 100755 --- a/dist/src/main/amoro-bin/conf/plugins/process-factories.yaml +++ b/dist/src/main/amoro-bin/conf/plugins/process-factories.yaml @@ -20,14 +20,17 @@ process-factories: enabled: true priority: 100 properties: - sync-table-meta.enabled: true - sync-table-meta.interval: 1 h - sync-table-meta.trigger-parallelism: 1 + sync-table-meta.enabled: false + sync-table-meta.interval: 48h + sync-table-meta.trigger-parallelism: 2 + expire-snapshots.enabled: true + expire-snapshots.interval: 24h + spark-version: 354 - name: paimon enabled: true priority: 100 properties: - paimon-optimizer.enabled: false + paimon-optimizer.enabled: true - name: iceberg enabled: true priority: 100 diff --git a/prev-conf/plugins/execute-engines.yaml b/prev-conf/plugins/execute-engines.yaml index c943ed110..230bc4608 100644 --- a/prev-conf/plugins/execute-engines.yaml +++ b/prev-conf/plugins/execute-engines.yaml @@ -26,5 +26,6 @@ execute-engines: enabled: true priority: 100 properties: - default.thread-count: 10 - table-meta-sync.thread-count: 3 + execute.user: amoro + default-spark-version: 354 + source-tag: AMORO diff --git a/prev-conf/plugins/process-factories.yaml b/prev-conf/plugins/process-factories.yaml index 514b14cf9..8e8cc6a50 100644 --- a/prev-conf/plugins/process-factories.yaml +++ b/prev-conf/plugins/process-factories.yaml @@ -16,7 +16,7 @@ # process-factories: - - name: paimon + - name: paimon-maintain enabled: true priority: 100 properties: @@ -25,5 +25,9 @@ process-factories: sync-table-meta.trigger-parallelism: 2 expire-snapshots.enabled: true expire-snapshots.interval: 24h - expire-snapshots.spark-version: 354 + spark-version: 354 + - name: paimon + enabled: true + priority: 100 + properties: paimon-optimizer.enabled: true
