This is an automated email from the ASF dual-hosted git repository.
zhoujinsong pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/amoro.git
The following commit(s) were added to refs/heads/master by this push:
new 0704a1ee9 [AMORO-4281]Fix cleanning process trigger issue for mixed
format tables (#4282)
0704a1ee9 is described below
commit 0704a1ee92df82bd923074d6714cb7612b45d964
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]>
---
.../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()));
+ }
+}
