Skip to content

Commit 743e6f1

Browse files
baiyangtxzhangyongxiang.alphaAime
authored
[Improvement]: Refactor process API and table runtime extension interfaces (#4097)
* improvement: refactor process API and table runtime extension interfaces # Context Split from upstream PR #4081 to make review easier. This MR introduces the new process extension APIs in common module, without touching most AMS internals yet. # Changes - Promote `ProcessFactory` to a richer abstraction in `amoro-common`, preparing for plugin-based table process extension. - Adjust `Action`/`IcebergActions` and `TableRuntime` to align with the new process model. - Move `ActionCoordinator` into `amoro-common` so it can be shared as a public abstraction. - Introduce `ProcessTriggerStrategy` to describe trigger policies for processes. - Clean up legacy process state classes which will be replaced by the new abstractions in follow-up MRs. # Notes - This commit only updates the common module and the shared `ActionCoordinator` API; AMS-side wiring and runtime refactors will be done in separate branches/MRs as discussed. Co-Authored-By: Aime <aime@bytedance.com> Change-Id: If84ada8fcae1cfb11577d56d3866db7ce0949102 * fix compile error * fix compile error * fix compile error3 * getTableConfig * refactor --------- Co-authored-by: zhangyongxiang.alpha <zhangyongxiang.alpha@bytedance.com> Co-authored-by: Aime <aime@bytedance.com>
1 parent 853562e commit 743e6f1

22 files changed

Lines changed: 285 additions & 948 deletions

amoro-ams/src/main/java/org/apache/amoro/server/persistence/mapper/ProcessStateMapper.java

Lines changed: 0 additions & 87 deletions
This file was deleted.

amoro-ams/src/main/java/org/apache/amoro/server/process/ActionCoordinatorScheduler.java

Lines changed: 5 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -20,13 +20,16 @@
2020

2121
import org.apache.amoro.TableFormat;
2222
import org.apache.amoro.TableRuntime;
23+
import org.apache.amoro.process.ActionCoordinator;
2324
import org.apache.amoro.process.TableProcess;
2425
import org.apache.amoro.process.TableProcessStore;
2526
import org.apache.amoro.server.scheduler.PeriodicTableScheduler;
2627
import org.apache.amoro.server.table.TableService;
2728
import org.slf4j.Logger;
2829
import org.slf4j.LoggerFactory;
2930

31+
import java.util.Optional;
32+
3033
/**
3134
* Periodic scheduler that delegates scheduling decisions to an {@link ActionCoordinator}. It
3235
* creates, recovers and retries table processes via {@link ProcessService}.
@@ -95,8 +98,8 @@ protected boolean enabled(TableRuntime tableRuntime) {
9598
*/
9699
@Override
97100
protected void execute(TableRuntime tableRuntime) {
98-
TableProcess process = coordinator.createTableProcess(tableRuntime);
99-
processService.register(tableRuntime, process);
101+
Optional<TableProcess> process = coordinator.trigger(tableRuntime);
102+
process.ifPresent(p -> processService.register(tableRuntime, p));
100103
}
101104

102105
/**
@@ -110,16 +113,6 @@ protected void recover(TableRuntime tableRuntime, TableProcessStore processStore
110113
processService.recover(tableRuntime, process);
111114
}
112115

113-
/**
114-
* Retry a failed table process.
115-
*
116-
* @param process process to retry
117-
*/
118-
protected void retry(TableProcess process) {
119-
process = coordinator.retryTableProcess(process);
120-
processService.retry(process);
121-
}
122-
123116
/**
124117
* Get executor delay from coordinator.
125118
*

amoro-ams/src/main/java/org/apache/amoro/server/process/ProcessService.java

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
import org.apache.amoro.TableRuntime;
2626
import org.apache.amoro.config.Configurations;
2727
import org.apache.amoro.config.TableConfiguration;
28+
import org.apache.amoro.process.ActionCoordinator;
2829
import org.apache.amoro.process.ProcessEvent;
2930
import org.apache.amoro.process.ProcessStatus;
3031
import org.apache.amoro.process.TableProcess;
@@ -242,7 +243,7 @@ private void executeOrTraceProcess(TableProcess process) {
242243
"Regular Retry.",
243244
process.getProcessParameters(),
244245
process.getSummary());
245-
scheduler.retry(process);
246+
executeOrTraceProcess(process);
246247
} else {
247248
untrackTableProcessInstance(
248249
process.getTableRuntime().getTableIdentifier(), process.getId());

amoro-ams/src/main/java/org/apache/amoro/server/process/TableProcessMeta.java

Lines changed: 0 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919
package org.apache.amoro.server.process;
2020

2121
import org.apache.amoro.process.ProcessStatus;
22-
import org.apache.amoro.process.TableProcessState;
2322
import org.apache.amoro.process.TableProcessStore;
2423

2524
import java.util.HashMap;
@@ -189,25 +188,6 @@ public static TableProcessMeta fromTableProcessStore(TableProcessStore tableProc
189188
return tableProcessMeta;
190189
}
191190

192-
@Deprecated
193-
public static TableProcessMeta fromTableProcessState(TableProcessState tableProcessState) {
194-
TableProcessMeta tableProcessMeta = new TableProcessMeta();
195-
tableProcessMeta.setProcessId(tableProcessState.getId());
196-
tableProcessMeta.setTableId(tableProcessState.getTableIdentifier().getId());
197-
tableProcessMeta.setExternalProcessIdentifier(tableProcessState.getExternalProcessIdentifier());
198-
tableProcessMeta.setStatus(tableProcessState.getStatus());
199-
tableProcessMeta.setProcessType(tableProcessState.getAction().getName());
200-
tableProcessMeta.setProcessStage(tableProcessState.getStage().getDesc());
201-
tableProcessMeta.setExecutionEngine(tableProcessState.getExecutionEngine());
202-
tableProcessMeta.setRetryNumber(tableProcessState.getRetryNumber());
203-
tableProcessMeta.setCreateTime(tableProcessState.getStartTime());
204-
tableProcessMeta.setFinishTime(tableProcessState.getEndTime());
205-
tableProcessMeta.setFailMessage(tableProcessState.getFailedReason());
206-
tableProcessMeta.setProcessParameters(tableProcessState.getProcessParameters());
207-
tableProcessMeta.setSummary(tableProcessState.getSummary());
208-
return tableProcessMeta;
209-
}
210-
211191
public static TableProcessMeta of(
212192
long processId,
213193
long tableId,

0 commit comments

Comments
 (0)