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 <[email protected]> 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 <[email protected]> (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())); + } +}
