Skip to content

Commit 24a094f

Browse files
张文领zhangwl9
张文领
authored and
zhangwl9
committed
fixup
1 parent 16c4bd6 commit 24a094f

3 files changed

Lines changed: 193 additions & 34 deletions

File tree

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

Lines changed: 38 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
import org.apache.amoro.server.utils.SnowflakeIdGenerator;
3737
import org.apache.amoro.shade.guava32.com.google.common.annotations.VisibleForTesting;
3838
import org.apache.amoro.shade.guava32.com.google.common.util.concurrent.ThreadFactoryBuilder;
39+
import org.apache.amoro.utils.ExceptionUtil;
3940
import org.apache.commons.lang3.StringUtils;
4041
import org.slf4j.Logger;
4142
import org.slf4j.LoggerFactory;
@@ -58,9 +59,9 @@ public abstract class PeriodicTableScheduler extends RuntimeHandlerChain {
5859
private static final String CLEANUP_EXECUTION_ENGINE = "AMORO";
5960
private static final String CLEANUP_PROCESS_STAGE = "CLEANUP";
6061
private static final String EXTERNAL_PROCESS_IDENTIFIER = "";
62+
private static final SnowflakeIdGenerator ID_GENERATOR = new SnowflakeIdGenerator();
6163

62-
private final SnowflakeIdGenerator idGenerator = new SnowflakeIdGenerator();
63-
private final PersistencyHelper persistencyHelper = new PersistencyHelper();
64+
private final PersistenceHelper persistenceHelper = new PersistenceHelper();
6465

6566
protected final Set<ServerTableIdentifier> scheduledTables =
6667
Collections.synchronizedSet(new HashSet<>());
@@ -152,6 +153,11 @@ private void executeTask(TableRuntime tableRuntime) {
152153
}
153154
} catch (Throwable t) {
154155
executionError = t;
156+
logger.error(
157+
"Failed to execute cleanup operation {} for table {}",
158+
cleanupOperation,
159+
tableRuntime.getTableIdentifier(),
160+
t);
155161
} finally {
156162
persistCleanupResult(tableRuntime, cleanupOperation, cleanProcessMeta, executionError);
157163

@@ -178,16 +184,38 @@ protected boolean shouldExecute(Long lastCleanupEndTime) {
178184
return true;
179185
}
180186

187+
private void persistUpdatingCleanupTime(TableRuntime tableRuntime) {
188+
CleanupOperation cleanupOperation = getCleanupOperation();
189+
if (shouldSkipOperation(tableRuntime, cleanupOperation)) {
190+
return;
191+
}
192+
193+
try {
194+
long currentTime = System.currentTimeMillis();
195+
((DefaultTableRuntime) tableRuntime).updateLastCleanTime(cleanupOperation, currentTime);
196+
197+
logger.debug(
198+
"Update lastCleanTime for table {} with cleanup operation {}",
199+
tableRuntime.getTableIdentifier().getTableName(),
200+
cleanupOperation);
201+
} catch (Exception e) {
202+
logger.error(
203+
"Failed to update lastCleanTime for table {}",
204+
tableRuntime.getTableIdentifier().getTableName(),
205+
e);
206+
}
207+
}
208+
181209
@VisibleForTesting
182-
public TableProcessMeta createCleanupProcessInfo(
210+
protected TableProcessMeta createCleanupProcessInfo(
183211
TableRuntime tableRuntime, CleanupOperation cleanupOperation) {
184212

185213
if (shouldSkipOperation(tableRuntime, cleanupOperation)) {
186214
return null;
187215
}
188216

189217
TableProcessMeta cleanProcessMeta = buildProcessMeta(tableRuntime, cleanupOperation);
190-
persistencyHelper.beginAndPersistCleanupProcess(cleanProcessMeta);
218+
persistenceHelper.beginAndPersistCleanupProcess(cleanProcessMeta);
191219

192220
logger.debug(
193221
"Successfully persist cleanup process [processId={}, tableId={}, processType={}]",
@@ -199,7 +227,7 @@ public TableProcessMeta createCleanupProcessInfo(
199227
}
200228

201229
@VisibleForTesting
202-
public void persistCleanupResult(
230+
protected void persistCleanupResult(
203231
TableRuntime tableRuntime,
204232
CleanupOperation cleanupOperation,
205233
TableProcessMeta cleanProcessMeta,
@@ -211,18 +239,13 @@ public void persistCleanupResult(
211239

212240
if (executionError != null) {
213241
cleanProcessMeta.setStatus(ProcessStatus.FAILED);
214-
String message = executionError.getMessage();
215-
if (message == null) {
216-
message = executionError.getClass().getName();
217-
}
218-
219-
cleanProcessMeta.setFailMessage(message);
242+
cleanProcessMeta.setFailMessage(ExceptionUtil.getErrorMessage(executionError, 4000));
220243
} else {
221244
cleanProcessMeta.setStatus(ProcessStatus.SUCCESS);
222245
}
223246

224247
long endTime = System.currentTimeMillis();
225-
persistencyHelper.persistAndSetCompleted(
248+
persistenceHelper.persistAndSetCompleted(
226249
tableRuntime, cleanupOperation, cleanProcessMeta, endTime);
227250

228251
logger.debug(
@@ -236,7 +259,7 @@ private TableProcessMeta buildProcessMeta(
236259

237260
TableProcessMeta cleanProcessMeta = new TableProcessMeta();
238261
cleanProcessMeta.setTableId(tableRuntime.getTableIdentifier().getId());
239-
cleanProcessMeta.setProcessId(idGenerator.generateId());
262+
cleanProcessMeta.setProcessId(ID_GENERATOR.generateId());
240263
cleanProcessMeta.setExternalProcessIdentifier(EXTERNAL_PROCESS_IDENTIFIER);
241264
cleanProcessMeta.setStatus(ProcessStatus.RUNNING);
242265
cleanProcessMeta.setProcessType(cleanupOperation.name());
@@ -250,9 +273,9 @@ private TableProcessMeta buildProcessMeta(
250273
return cleanProcessMeta;
251274
}
252275

253-
private static class PersistencyHelper extends PersistentBase {
276+
private static class PersistenceHelper extends PersistentBase {
254277

255-
public PersistencyHelper() {}
278+
public PersistenceHelper() {}
256279

257280
private void beginAndPersistCleanupProcess(TableProcessMeta meta) {
258281
doAsTransaction(

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

Lines changed: 36 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,23 +24,36 @@
2424
import org.apache.amoro.server.table.TableService;
2525
import org.apache.amoro.server.table.cleanup.CleanupOperation;
2626

27+
import java.lang.reflect.InvocationTargetException;
28+
import java.lang.reflect.Method;
29+
2730
/**
2831
* Test table executor implementation for testing PeriodicTableScheduler functionality. This class
2932
* allows configuration of cleanup operations and enabled state for testing purposes.
3033
*/
3134
class PeriodicTableSchedulerTestBase extends PeriodicTableScheduler {
3235
private final CleanupOperation cleanupOperation;
3336
private final boolean enabled;
37+
private final RuntimeException executionException;
3438
private static final long SNAPSHOTS_EXPIRING_INTERVAL = 60 * 60 * 1000L; // 1 hour
3539
private static final long ORPHAN_FILES_CLEANING_INTERVAL = 24 * 60 * 60 * 1000L; // 1 day
3640
private static final long DANGLING_DELETE_FILES_CLEANING_INTERVAL = 24 * 60 * 60 * 1000L;
3741
private static final long DATA_EXPIRING_INTERVAL = 60 * 60 * 1000L; // 1 hour
3842

3943
public PeriodicTableSchedulerTestBase(
4044
TableService tableService, CleanupOperation cleanupOperation, boolean enabled) {
45+
this(tableService, cleanupOperation, enabled, null);
46+
}
47+
48+
public PeriodicTableSchedulerTestBase(
49+
TableService tableService,
50+
CleanupOperation cleanupOperation,
51+
boolean enabled,
52+
RuntimeException executionException) {
4153
super(tableService, 1);
4254
this.cleanupOperation = cleanupOperation;
4355
this.enabled = enabled;
56+
this.executionException = executionException;
4457
}
4558

4659
@Override
@@ -60,7 +73,9 @@ protected boolean enabled(TableRuntime tableRuntime) {
6073

6174
@Override
6275
protected void execute(TableRuntime tableRuntime) {
63-
// Do nothing in test
76+
if (executionException != null) {
77+
throw executionException;
78+
}
6479
}
6580

6681
@Override
@@ -102,4 +117,24 @@ public void persistCleanupResultForTest(
102117
Throwable error) {
103118
persistCleanupResult(tableRuntime, operation, meta, error);
104119
}
120+
121+
public void executeTaskForTest(TableRuntime tableRuntime) {
122+
try {
123+
Method executeTask =
124+
PeriodicTableScheduler.class.getDeclaredMethod("executeTask", TableRuntime.class);
125+
executeTask.setAccessible(true);
126+
executeTask.invoke(this, tableRuntime);
127+
} catch (InvocationTargetException e) {
128+
Throwable cause = e.getCause();
129+
if (cause instanceof RuntimeException) {
130+
throw (RuntimeException) cause;
131+
}
132+
if (cause instanceof Error) {
133+
throw (Error) cause;
134+
}
135+
throw new RuntimeException(cause);
136+
} catch (ReflectiveOperationException e) {
137+
throw new RuntimeException("Failed to invoke executeTask for test", e);
138+
}
139+
}
105140
}

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

Lines changed: 119 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -18,11 +18,13 @@
1818

1919
package org.apache.amoro.server.scheduler.inline;
2020

21+
import org.apache.amoro.AmoroTable;
2122
import org.apache.amoro.ServerTableIdentifier;
2223
import org.apache.amoro.TableFormat;
2324
import org.apache.amoro.TableRuntime;
2425
import org.apache.amoro.config.TableConfiguration;
2526
import org.apache.amoro.process.ProcessStatus;
27+
import org.apache.amoro.server.catalog.InternalCatalog;
2628
import org.apache.amoro.server.persistence.PersistentBase;
2729
import org.apache.amoro.server.persistence.TableRuntimeMeta;
2830
import org.apache.amoro.server.persistence.mapper.TableMetaMapper;
@@ -31,7 +33,8 @@
3133
import org.apache.amoro.server.process.TableProcessMeta;
3234
import org.apache.amoro.server.table.DefaultTableRuntime;
3335
import org.apache.amoro.server.table.DefaultTableRuntimeStore;
34-
import org.apache.amoro.server.table.TableRuntimeHandler;
36+
import org.apache.amoro.server.table.RuntimeHandlerChain;
37+
import org.apache.amoro.server.table.TableService;
3538
import org.apache.amoro.server.table.cleanup.CleanupOperation;
3639
import org.apache.amoro.table.TableRuntimeStore;
3740
import org.apache.amoro.table.TableSummary;
@@ -41,6 +44,7 @@
4144
import java.util.Arrays;
4245
import java.util.Collections;
4346
import java.util.List;
47+
import java.util.concurrent.atomic.AtomicInteger;
4448

4549
/**
4650
* This class tests all aspects of cleanup operation handling in {@link
@@ -60,17 +64,51 @@ public class TestPeriodicTableSchedulerCleanup extends PersistentBase {
6064
}
6165
}
6266

63-
private static final TableRuntimeHandler TEST_HANDLER =
64-
new TableRuntimeHandler() {
65-
@Override
66-
public void handleTableChanged(
67-
TableRuntime tableRuntime,
68-
org.apache.amoro.server.optimizing.OptimizingStatus originalStatus) {}
67+
private static class OneShotTableService implements TableService {
68+
private final TableRuntime tableRuntime;
69+
private final AtomicInteger getRuntimeCalls = new AtomicInteger();
6970

70-
@Override
71-
public void handleTableChanged(
72-
TableRuntime tableRuntime, TableConfiguration originalConfig) {}
73-
};
71+
private OneShotTableService(TableRuntime tableRuntime) {
72+
this.tableRuntime = tableRuntime;
73+
}
74+
75+
@Override
76+
public void addHandlerChain(RuntimeHandlerChain handler) {}
77+
78+
@Override
79+
public void initialize() {}
80+
81+
@Override
82+
public void dispose() {}
83+
84+
@Override
85+
public void onTableCreated(InternalCatalog catalog, ServerTableIdentifier identifier) {}
86+
87+
@Override
88+
public void onTableDropped(InternalCatalog catalog, ServerTableIdentifier identifier) {}
89+
90+
@Override
91+
public TableRuntime getRuntime(Long tableId) {
92+
return getRuntimeCalls.getAndIncrement() == 0
93+
&& tableRuntime.getTableIdentifier().getId() == tableId
94+
? tableRuntime
95+
: null;
96+
}
97+
98+
@Override
99+
public AmoroTable<?> loadTable(ServerTableIdentifier identifier) {
100+
throw new UnsupportedOperationException(
101+
"loadTable is not expected in cleanup scheduler tests");
102+
}
103+
104+
@Override
105+
public void handleTableChanged(
106+
TableRuntime tableRuntime,
107+
org.apache.amoro.server.optimizing.OptimizingStatus originalStatus) {}
108+
109+
@Override
110+
public void handleTableChanged(TableRuntime tableRuntime, TableConfiguration originalConfig) {}
111+
}
74112

75113
/**
76114
* Create a test server table identifier with the given ID
@@ -153,6 +191,15 @@ private PeriodicTableSchedulerTestBase createTestExecutor(
153191
return new PeriodicTableSchedulerTestBase(null, cleanupOperation, enabled);
154192
}
155193

194+
private PeriodicTableSchedulerTestBase createTestExecutor(
195+
TableService tableService,
196+
CleanupOperation cleanupOperation,
197+
boolean enabled,
198+
RuntimeException executionException) {
199+
return new PeriodicTableSchedulerTestBase(
200+
tableService, cleanupOperation, enabled, executionException);
201+
}
202+
156203
/**
157204
* Create a test table executor with default enabled state (true)
158205
*
@@ -299,7 +346,7 @@ public void testCleanupProcessPersistence() {
299346
executor.persistCleanupResultForTest(tableRuntime, operation, processMeta.copy(), null);
300347
TableProcessMeta persistedSuccess = getProcessMeta(processMeta.getProcessId());
301348
Assert.assertEquals(ProcessStatus.SUCCESS, persistedSuccess.getStatus());
302-
Assert.assertTrue(persistedSuccess.getCreateTime() < persistedSuccess.getFinishTime());
349+
Assert.assertTrue(persistedSuccess.getCreateTime() <= persistedSuccess.getFinishTime());
303350

304351
// Scenario 3: Failure handling - verify error persistence
305352
processMeta = executor.createCleanupProcessInfoForTest(tableRuntime, operation);
@@ -308,8 +355,63 @@ public void testCleanupProcessPersistence() {
308355

309356
TableProcessMeta persistedFailure = getProcessMeta(processMeta.getProcessId());
310357
Assert.assertEquals(ProcessStatus.FAILED, persistedFailure.getStatus());
311-
Assert.assertEquals(testError.getMessage(), persistedFailure.getFailMessage());
358+
Assert.assertTrue(persistedFailure.getFailMessage().contains(testError.getMessage()));
359+
Assert.assertTrue(persistedFailure.getFailMessage().length() <= 4000);
360+
361+
cleanUpTableProcess(tableId);
362+
}
363+
}
364+
365+
@Test
366+
public void testExecuteTaskPersistsSuccess() {
367+
long tableId = 300L;
368+
CleanupOperation operation = CleanupOperation.ORPHAN_FILES_CLEANING;
369+
prepareTestEnvironment(Collections.singletonList(tableId));
370+
371+
ServerTableIdentifier identifier = createTableIdentifier(tableId);
372+
DefaultTableRuntime tableRuntime = createDefaultTableRuntime(identifier);
373+
PeriodicTableSchedulerTestBase executor =
374+
createTestExecutor(new OneShotTableService(tableRuntime), operation, true, null);
312375

376+
try {
377+
executor.executeTaskForTest(tableRuntime);
378+
379+
List<TableProcessMeta> processMetas =
380+
getAs(TableProcessMapper.class, m -> m.listProcessMeta(tableId, operation.name(), null));
381+
Assert.assertEquals(1, processMetas.size());
382+
Assert.assertEquals(ProcessStatus.SUCCESS, processMetas.get(0).getStatus());
383+
Assert.assertTrue(processMetas.get(0).getFinishTime() >= processMetas.get(0).getCreateTime());
384+
} finally {
385+
executor.gracefulShutdown();
386+
cleanUpTableProcess(tableId);
387+
}
388+
}
389+
390+
@Test
391+
public void testExecuteTaskPersistsFailureWhenExecutionThrows() {
392+
long tableId = 301L;
393+
CleanupOperation operation = CleanupOperation.DATA_EXPIRING;
394+
prepareTestEnvironment(Collections.singletonList(tableId));
395+
396+
ServerTableIdentifier identifier = createTableIdentifier(tableId);
397+
DefaultTableRuntime tableRuntime = createDefaultTableRuntime(identifier);
398+
RuntimeException executionException = new RuntimeException("Execute task failed");
399+
PeriodicTableSchedulerTestBase executor =
400+
createTestExecutor(
401+
new OneShotTableService(tableRuntime), operation, true, executionException);
402+
403+
try {
404+
executor.executeTaskForTest(tableRuntime);
405+
406+
List<TableProcessMeta> processMetas =
407+
getAs(TableProcessMapper.class, m -> m.listProcessMeta(tableId, operation.name(), null));
408+
Assert.assertEquals(1, processMetas.size());
409+
Assert.assertEquals(ProcessStatus.FAILED, processMetas.get(0).getStatus());
410+
Assert.assertTrue(
411+
processMetas.get(0).getFailMessage().contains(executionException.getMessage()));
412+
Assert.assertTrue(processMetas.get(0).getFailMessage().length() <= 4000);
413+
} finally {
414+
executor.gracefulShutdown();
313415
cleanUpTableProcess(tableId);
314416
}
315417
}
@@ -322,10 +424,9 @@ private TableProcessMeta getProcessMeta(long processId) {
322424
private void cleanUpTableProcess(long tableId) {
323425
doAs(
324426
TableProcessMapper.class,
325-
mapper -> {
326-
mapper
327-
.listProcessMeta(tableId, null, null)
328-
.forEach(p -> mapper.deleteBefore(tableId, p.getProcessId()));
329-
});
427+
mapper ->
428+
mapper
429+
.listProcessMeta(tableId, null, null)
430+
.forEach(p -> mapper.deleteBefore(tableId, p.getProcessId())));
330431
}
331432
}

0 commit comments

Comments
 (0)