Skip to content

Commit 16c4bd6

Browse files
张文领zhangwl9
张文领
authored and
zhangwl9
committed
Saving cleanup info in table_process
# Conflicts: # amoro-ams/src/main/java/org/apache/amoro/server/scheduler/PeriodicTableScheduler.java
1 parent f0dc274 commit 16c4bd6

3 files changed

Lines changed: 219 additions & 16 deletions

File tree

‎amoro-ams/src/main/java/org/apache/amoro/server/scheduler/PeriodicTableScheduler.java‎

Lines changed: 139 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -24,17 +24,24 @@
2424
import org.apache.amoro.ServerTableIdentifier;
2525
import org.apache.amoro.TableRuntime;
2626
import org.apache.amoro.config.TableConfiguration;
27+
import org.apache.amoro.process.ProcessStatus;
2728
import org.apache.amoro.server.optimizing.OptimizingStatus;
29+
import org.apache.amoro.server.persistence.PersistentBase;
30+
import org.apache.amoro.server.persistence.mapper.TableProcessMapper;
31+
import org.apache.amoro.server.process.TableProcessMeta;
2832
import org.apache.amoro.server.table.DefaultTableRuntime;
2933
import org.apache.amoro.server.table.RuntimeHandlerChain;
3034
import org.apache.amoro.server.table.TableService;
3135
import org.apache.amoro.server.table.cleanup.CleanupOperation;
36+
import org.apache.amoro.server.utils.SnowflakeIdGenerator;
37+
import org.apache.amoro.shade.guava32.com.google.common.annotations.VisibleForTesting;
3238
import org.apache.amoro.shade.guava32.com.google.common.util.concurrent.ThreadFactoryBuilder;
3339
import org.apache.commons.lang3.StringUtils;
3440
import org.slf4j.Logger;
3541
import org.slf4j.LoggerFactory;
3642

3743
import java.util.Collections;
44+
import java.util.HashMap;
3845
import java.util.HashSet;
3946
import java.util.List;
4047
import java.util.Locale;
@@ -48,6 +55,12 @@ public abstract class PeriodicTableScheduler extends RuntimeHandlerChain {
4855
protected final Logger logger = LoggerFactory.getLogger(getClass());
4956

5057
private static final long START_DELAY = 10 * 1000L;
58+
private static final String CLEANUP_EXECUTION_ENGINE = "AMORO";
59+
private static final String CLEANUP_PROCESS_STAGE = "CLEANUP";
60+
private static final String EXTERNAL_PROCESS_IDENTIFIER = "";
61+
62+
private final SnowflakeIdGenerator idGenerator = new SnowflakeIdGenerator();
63+
private final PersistencyHelper persistencyHelper = new PersistencyHelper();
5164

5265
protected final Set<ServerTableIdentifier> scheduledTables =
5366
Collections.synchronizedSet(new HashSet<>());
@@ -123,16 +136,25 @@ private void scheduleTableExecution(TableRuntime tableRuntime, long delay) {
123136
}
124137

125138
private void executeTask(TableRuntime tableRuntime) {
139+
TableProcessMeta cleanProcessMeta = null;
140+
CleanupOperation cleanupOperation = null;
141+
Throwable executionError = null;
142+
126143
try {
127144
if (isExecutable(tableRuntime)) {
145+
cleanupOperation = getCleanupOperation();
146+
cleanProcessMeta = createCleanupProcessInfo(tableRuntime, cleanupOperation);
147+
128148
execute(tableRuntime);
129149
// Different tables take different amounts of time to execute the end of execute(),
130150
// so you need to perform the update operation separately for each table.
131151
persistUpdatingCleanupTime(tableRuntime);
132152
}
133-
} catch (Exception e) {
134-
logger.error("exception when schedule for table: {}", tableRuntime.getTableIdentifier(), e);
153+
} catch (Throwable t) {
154+
executionError = t;
135155
} finally {
156+
persistCleanupResult(tableRuntime, cleanupOperation, cleanProcessMeta, executionError);
157+
136158
scheduledTables.remove(tableRuntime.getTableIdentifier());
137159
scheduleIfNecessary(tableRuntime, getNextExecutingTime(tableRuntime));
138160
}
@@ -156,25 +178,126 @@ protected boolean shouldExecute(Long lastCleanupEndTime) {
156178
return true;
157179
}
158180

159-
private void persistUpdatingCleanupTime(TableRuntime tableRuntime) {
160-
CleanupOperation cleanupOperation = getCleanupOperation();
181+
@VisibleForTesting
182+
public TableProcessMeta createCleanupProcessInfo(
183+
TableRuntime tableRuntime, CleanupOperation cleanupOperation) {
184+
161185
if (shouldSkipOperation(tableRuntime, cleanupOperation)) {
186+
return null;
187+
}
188+
189+
TableProcessMeta cleanProcessMeta = buildProcessMeta(tableRuntime, cleanupOperation);
190+
persistencyHelper.beginAndPersistCleanupProcess(cleanProcessMeta);
191+
192+
logger.debug(
193+
"Successfully persist cleanup process [processId={}, tableId={}, processType={}]",
194+
cleanProcessMeta.getProcessId(),
195+
cleanProcessMeta.getTableId(),
196+
cleanProcessMeta.getProcessType());
197+
198+
return cleanProcessMeta;
199+
}
200+
201+
@VisibleForTesting
202+
public void persistCleanupResult(
203+
TableRuntime tableRuntime,
204+
CleanupOperation cleanupOperation,
205+
TableProcessMeta cleanProcessMeta,
206+
Throwable executionError) {
207+
208+
if (cleanProcessMeta == null) {
162209
return;
163210
}
164211

165-
try {
166-
long currentTime = System.currentTimeMillis();
167-
((DefaultTableRuntime) tableRuntime).updateLastCleanTime(cleanupOperation, currentTime);
212+
if (executionError != null) {
213+
cleanProcessMeta.setStatus(ProcessStatus.FAILED);
214+
String message = executionError.getMessage();
215+
if (message == null) {
216+
message = executionError.getClass().getName();
217+
}
168218

169-
logger.debug(
170-
"Update lastCleanTime for table {} with cleanup operation {}",
171-
tableRuntime.getTableIdentifier().getTableName(),
172-
cleanupOperation);
173-
} catch (Exception e) {
174-
logger.error(
175-
"Failed to update lastCleanTime for table {}",
176-
tableRuntime.getTableIdentifier().getTableName(),
177-
e);
219+
cleanProcessMeta.setFailMessage(message);
220+
} else {
221+
cleanProcessMeta.setStatus(ProcessStatus.SUCCESS);
222+
}
223+
224+
long endTime = System.currentTimeMillis();
225+
persistencyHelper.persistAndSetCompleted(
226+
tableRuntime, cleanupOperation, cleanProcessMeta, endTime);
227+
228+
logger.debug(
229+
"Successfully updated lastCleanTime and cleanupProcess for table {} with cleanup operation {}",
230+
tableRuntime.getTableIdentifier().getTableName(),
231+
cleanupOperation);
232+
}
233+
234+
private TableProcessMeta buildProcessMeta(
235+
TableRuntime tableRuntime, CleanupOperation cleanupOperation) {
236+
237+
TableProcessMeta cleanProcessMeta = new TableProcessMeta();
238+
cleanProcessMeta.setTableId(tableRuntime.getTableIdentifier().getId());
239+
cleanProcessMeta.setProcessId(idGenerator.generateId());
240+
cleanProcessMeta.setExternalProcessIdentifier(EXTERNAL_PROCESS_IDENTIFIER);
241+
cleanProcessMeta.setStatus(ProcessStatus.RUNNING);
242+
cleanProcessMeta.setProcessType(cleanupOperation.name());
243+
cleanProcessMeta.setProcessStage(CLEANUP_PROCESS_STAGE);
244+
cleanProcessMeta.setExecutionEngine(CLEANUP_EXECUTION_ENGINE);
245+
cleanProcessMeta.setRetryNumber(0);
246+
cleanProcessMeta.setCreateTime(System.currentTimeMillis());
247+
cleanProcessMeta.setProcessParameters(new HashMap<>());
248+
cleanProcessMeta.setSummary(new HashMap<>());
249+
250+
return cleanProcessMeta;
251+
}
252+
253+
private static class PersistencyHelper extends PersistentBase {
254+
255+
public PersistencyHelper() {}
256+
257+
private void beginAndPersistCleanupProcess(TableProcessMeta meta) {
258+
doAsTransaction(
259+
() ->
260+
doAs(
261+
TableProcessMapper.class,
262+
mapper ->
263+
mapper.insertProcess(
264+
meta.getTableId(),
265+
meta.getProcessId(),
266+
meta.getExternalProcessIdentifier(),
267+
meta.getStatus(),
268+
meta.getProcessType(),
269+
meta.getProcessStage(),
270+
meta.getExecutionEngine(),
271+
meta.getRetryNumber(),
272+
meta.getCreateTime(),
273+
meta.getProcessParameters(),
274+
meta.getSummary())));
275+
}
276+
277+
private void persistAndSetCompleted(
278+
TableRuntime tableRuntime,
279+
CleanupOperation cleanupOperation,
280+
TableProcessMeta meta,
281+
long endTime) {
282+
283+
doAsTransaction(
284+
() ->
285+
doAs(
286+
TableProcessMapper.class,
287+
mapper ->
288+
mapper.updateProcess(
289+
meta.getTableId(),
290+
meta.getProcessId(),
291+
meta.getExternalProcessIdentifier(),
292+
meta.getStatus(),
293+
meta.getProcessStage(),
294+
meta.getRetryNumber(),
295+
endTime,
296+
meta.getFailMessage(),
297+
meta.getProcessParameters(),
298+
meta.getSummary())),
299+
() ->
300+
((DefaultTableRuntime) tableRuntime).updateLastCleanTime(cleanupOperation, endTime));
178301
}
179302
}
180303

‎amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/PeriodicTableSchedulerTestBase.java‎

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

2121
import org.apache.amoro.TableRuntime;
22+
import org.apache.amoro.server.process.TableProcessMeta;
2223
import org.apache.amoro.server.scheduler.PeriodicTableScheduler;
2324
import org.apache.amoro.server.table.TableService;
2425
import org.apache.amoro.server.table.cleanup.CleanupOperation;
@@ -88,4 +89,17 @@ public boolean shouldExecuteTaskForTest(
8889
TableRuntime tableRuntime, CleanupOperation cleanupOperation) {
8990
return shouldExecuteTask(tableRuntime, cleanupOperation);
9091
}
92+
93+
public TableProcessMeta createCleanupProcessInfoForTest(
94+
TableRuntime tableRuntime, CleanupOperation operation) {
95+
return createCleanupProcessInfo(tableRuntime, operation);
96+
}
97+
98+
public void persistCleanupResultForTest(
99+
TableRuntime tableRuntime,
100+
CleanupOperation operation,
101+
TableProcessMeta meta,
102+
Throwable error) {
103+
persistCleanupResult(tableRuntime, operation, meta, error);
104+
}
91105
}

‎amoro-ams/src/test/java/org/apache/amoro/server/scheduler/inline/TestPeriodicTableSchedulerCleanup.java‎

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,10 +22,13 @@
2222
import org.apache.amoro.TableFormat;
2323
import org.apache.amoro.TableRuntime;
2424
import org.apache.amoro.config.TableConfiguration;
25+
import org.apache.amoro.process.ProcessStatus;
2526
import org.apache.amoro.server.persistence.PersistentBase;
2627
import org.apache.amoro.server.persistence.TableRuntimeMeta;
2728
import org.apache.amoro.server.persistence.mapper.TableMetaMapper;
29+
import org.apache.amoro.server.persistence.mapper.TableProcessMapper;
2830
import org.apache.amoro.server.persistence.mapper.TableRuntimeMapper;
31+
import org.apache.amoro.server.process.TableProcessMeta;
2932
import org.apache.amoro.server.table.DefaultTableRuntime;
3033
import org.apache.amoro.server.table.DefaultTableRuntimeStore;
3134
import org.apache.amoro.server.table.TableRuntimeHandler;
@@ -262,4 +265,67 @@ public void testShouldExecuteTaskWithNoneOperation() {
262265
boolean shouldExecute = executor.shouldExecuteTaskForTest(tableRuntime, CleanupOperation.NONE);
263266
Assert.assertTrue("Should always execute with NONE operation", shouldExecute);
264267
}
268+
269+
/**
270+
* Validates that process info is correctly saved to table_process with proper status transitions.
271+
*/
272+
@Test
273+
public void testCleanupProcessPersistence() {
274+
long baseTableId = 200L;
275+
List<CleanupOperation> operations =
276+
Arrays.asList(
277+
CleanupOperation.ORPHAN_FILES_CLEANING,
278+
CleanupOperation.DANGLING_DELETE_FILES_CLEANING,
279+
CleanupOperation.DATA_EXPIRING,
280+
CleanupOperation.SNAPSHOTS_EXPIRING);
281+
282+
for (int i = 0; i < operations.size(); i++) {
283+
long tableId = baseTableId + i;
284+
CleanupOperation operation = operations.get(i);
285+
prepareTestEnvironment(Collections.singletonList(tableId));
286+
287+
PeriodicTableSchedulerTestBase executor = createTestExecutor(operation);
288+
ServerTableIdentifier identifier = createTableIdentifier(tableId);
289+
DefaultTableRuntime tableRuntime = createDefaultTableRuntime(identifier);
290+
291+
// Scenario 1: Create process - verify initial persisted state
292+
TableProcessMeta processMeta =
293+
executor.createCleanupProcessInfoForTest(tableRuntime, operation);
294+
TableProcessMeta persistedRunning = getProcessMeta(processMeta.getProcessId());
295+
296+
Assert.assertEquals(ProcessStatus.RUNNING, persistedRunning.getStatus());
297+
298+
// Scenario 2: Success completion - verify status update and timestamps
299+
executor.persistCleanupResultForTest(tableRuntime, operation, processMeta.copy(), null);
300+
TableProcessMeta persistedSuccess = getProcessMeta(processMeta.getProcessId());
301+
Assert.assertEquals(ProcessStatus.SUCCESS, persistedSuccess.getStatus());
302+
Assert.assertTrue(persistedSuccess.getCreateTime() < persistedSuccess.getFinishTime());
303+
304+
// Scenario 3: Failure handling - verify error persistence
305+
processMeta = executor.createCleanupProcessInfoForTest(tableRuntime, operation);
306+
RuntimeException testError = new RuntimeException("Cleanup failed for " + operation);
307+
executor.persistCleanupResultForTest(tableRuntime, operation, processMeta.copy(), testError);
308+
309+
TableProcessMeta persistedFailure = getProcessMeta(processMeta.getProcessId());
310+
Assert.assertEquals(ProcessStatus.FAILED, persistedFailure.getStatus());
311+
Assert.assertEquals(testError.getMessage(), persistedFailure.getFailMessage());
312+
313+
cleanUpTableProcess(tableId);
314+
}
315+
}
316+
317+
private TableProcessMeta getProcessMeta(long processId) {
318+
return getAs(TableProcessMapper.class, m -> m.getProcessMeta(processId));
319+
}
320+
321+
/** Clean up table_process records for a specific table */
322+
private void cleanUpTableProcess(long tableId) {
323+
doAs(
324+
TableProcessMapper.class,
325+
mapper -> {
326+
mapper
327+
.listProcessMeta(tableId, null, null)
328+
.forEach(p -> mapper.deleteBefore(tableId, p.getProcessId()));
329+
});
330+
}
265331
}

0 commit comments

Comments
 (0)