Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

[ARCTIC-1737] 0.3.x fix the tasks of ScheduledExecutorService that halts when an exception or error is thrown #1743

Merged
merged 1 commit into from
Jul 26, 2023
Merged

[ARCTIC-1737] 0.3.x fix the tasks of ScheduledExecutorService that halts when an exception or error is thrown #1743

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 8 additions & 4 deletions ams/ams-server/src/main/java/com/netease/arctic/ams/server/optimize/OptimizeService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,11 +112,15 @@ private void init() {

@Override
public synchronized void checkOptimizeCheckTasks(long checkInterval) {
this.optimizeStatusCheckInterval = checkInterval;
if (checkTasks == null) {
checkTasks = new ScheduledTasks<>(ThreadPool.Type.OPTIMIZE_CHECK);
try {
this.optimizeStatusCheckInterval = checkInterval;
if (checkTasks == null) {
checkTasks = new ScheduledTasks<>(ThreadPool.Type.OPTIMIZE_CHECK);
}
internalCheckOptimizeChecker(checkInterval);
} catch (Throwable t) {
LOG.error("Failed to check optimize check tasks", t);
}
internalCheckOptimizeChecker(checkInterval);
}

private void internalCheckOptimizeChecker(long checkInterval) {
Expand Down
27 changes: 17 additions & 10 deletions ...rver/src/main/java/com/netease/arctic/ams/server/service/impl/OptimizeExecuteService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,8 @@
import com.netease.arctic.ams.server.service.ServiceContainer;
import com.netease.arctic.optimizer.Optimizer;
import com.netease.arctic.optimizer.factory.OptimizerFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;
import java.io.IOException;
Expand Down Expand Up @@ -111,23 +113,28 @@ public OptimizerFactory findOptimizerFactory(String type) throws NoSuchObjectExc

@Override
public void close() throws IOException {

}

public static class OptimizerMonitor {
private static final Logger LOG = LoggerFactory.getLogger(OptimizerMonitor.class);

private static final long OPTIMIZER_JOB_TIMEOUT = 10 * 60 * 1000;

public void monitorStatus() {
long currentTime = System.currentTimeMillis();
List<com.netease.arctic.ams.server.model.Optimizer> optimizers =
ServiceContainer.getOptimizerService().getOptimizers();
optimizers.forEach(optimizer -> {
if ((currentTime - optimizer.getUpdateTime().getTime()) > OPTIMIZER_JOB_TIMEOUT) {
ServiceContainer.getOptimizerService()
.updateOptimizerStatus(optimizer.getJobId(), TableTaskStatus.FAILED);
}
});
try {
long currentTime = System.currentTimeMillis();
List<com.netease.arctic.ams.server.model.Optimizer> optimizers =
ServiceContainer.getOptimizerService().getOptimizers();
optimizers.forEach(optimizer -> {
if ((currentTime - optimizer.getUpdateTime().getTime()) > OPTIMIZER_JOB_TIMEOUT) {
ServiceContainer.getOptimizerService()
.updateOptimizerStatus(optimizer.getJobId(), TableTaskStatus.FAILED);
}
});
} catch (Throwable t) {
LOG.error("monitor optimizer status failed", t);
}
}
}
}
Expand Down
26 changes: 15 additions & 11 deletions ...ver/src/main/java/com/netease/arctic/ams/server/service/impl/OrphanFilesCleanService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -69,18 +69,22 @@ public class OrphanFilesCleanService implements IOrphanFilesCleanService {

@Override
public synchronized void checkOrphanFilesCleanTasks() {
LOG.info("Schedule Orphan Cleaner");
if (cleanTasks == null) {
cleanTasks = new ScheduledTasks<>(ThreadPool.Type.ORPHAN);
try {
LOG.info("Schedule Orphan Cleaner");
if (cleanTasks == null) {
cleanTasks = new ScheduledTasks<>(ThreadPool.Type.ORPHAN);
}
List<TableMetadata> tables = ServiceContainer.getMetaService().listTables();
Set<TableIdentifier> ids = tables.stream().map(TableMetadata::getTableIdentifier).collect(Collectors.toSet());
cleanTasks.checkRunningTask(ids,
() -> 0L,
() -> CHECK_INTERVAL,
TableOrphanFileClean::new,
true);
LOG.info("Schedule Orphan Cleaner finished with {} tasks", ids.size());
} catch (Throwable t) {
LOG.error("unexpected error when checkOrphanFilesCleanTasks", t);
}
List<TableMetadata> tables = ServiceContainer.getMetaService().listTables();
Set<TableIdentifier> ids = tables.stream().map(TableMetadata::getTableIdentifier).collect(Collectors.toSet());
cleanTasks.checkRunningTask(ids,
() -> 0L,
() -> CHECK_INTERVAL,
TableOrphanFileClean::new,
true);
LOG.info("Schedule Orphan Cleaner finished with {} tasks", ids.size());
}

@Override
Expand Down
66 changes: 35 additions & 31 deletions ...er/src/main/java/com/netease/arctic/ams/server/service/impl/RuntimeDataExpireService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,39 +56,43 @@ public RuntimeDataExpireService() {
}

public void doExpire() {
List<TableMetadata> tableMetadata = metaService.listTables();
// expire and clear transaction table
tableMetadata.forEach(meta -> {
TableIdentifier identifier = meta.getTableIdentifier();
transactionService.expire(
identifier.buildTableIdentifier(),
System.currentTimeMillis() - this.txDataExpireInterval);
});
try {
List<TableMetadata> tableMetadata = metaService.listTables();
// expire and clear transaction table
tableMetadata.forEach(meta -> {
TableIdentifier identifier = meta.getTableIdentifier();
transactionService.expire(
identifier.buildTableIdentifier(),
System.currentTimeMillis() - this.txDataExpireInterval);
});

// expire and clear table_task_history table
tableMetadata.forEach(meta -> {
TableIdentifier identifier = meta.getTableIdentifier();
try {
TableOptimizeRuntime tableOptimizeRuntime =
optimizeService.getTableOptimizeItem(identifier).getTableOptimizeRuntime();
tableTaskHistoryService.expireTaskHistory(identifier,
tableOptimizeRuntime.getLatestTaskPlanGroup(),
System.currentTimeMillis() - this.taskHistoryDataExpireInterval);
} catch (Exception e) {
LOG.error("failed to expire and clear table_task_history table", e);
}
});
// expire and clear table_task_history table
tableMetadata.forEach(meta -> {
TableIdentifier identifier = meta.getTableIdentifier();
try {
TableOptimizeRuntime tableOptimizeRuntime =
optimizeService.getTableOptimizeItem(identifier).getTableOptimizeRuntime();
tableTaskHistoryService.expireTaskHistory(identifier,
tableOptimizeRuntime.getLatestTaskPlanGroup(),
System.currentTimeMillis() - this.taskHistoryDataExpireInterval);
} catch (Exception e) {
LOG.error("failed to expire and clear table_task_history table", e);
}
});

// expire and clear optimize_history table
tableMetadata.forEach(meta -> {
TableIdentifier identifier = meta.getTableIdentifier();
try {
optimizeService.expireOptimizeHistory(identifier,
System.currentTimeMillis() - this.optimizeHistoryDataExpireInterval);
} catch (Exception e) {
LOG.error("failed to expire and clear optimize_history table", e);
}
});
// expire and clear optimize_history table
tableMetadata.forEach(meta -> {
TableIdentifier identifier = meta.getTableIdentifier();
try {
optimizeService.expireOptimizeHistory(identifier,
System.currentTimeMillis() - this.optimizeHistoryDataExpireInterval);
} catch (Exception e) {
LOG.error("failed to expire and clear optimize_history table", e);
}
});
} catch (Throwable t) {
LOG.error("failed to expire and clear runtime data", t);
}
}

@Override
Expand Down
28 changes: 16 additions & 12 deletions ...rver/src/main/java/com/netease/arctic/ams/server/service/impl/SupportHiveSyncService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -64,19 +64,23 @@ public class SupportHiveSyncService implements ISupportHiveSyncService {

@Override
public void checkHiveSyncTasks() {
LOG.info("Schedule Support Hive Sync");
if (syncTasks == null) {
syncTasks = new ScheduledTasks<>(ThreadPool.Type.HIVE_SYNC);
try {
LOG.info("Schedule Support Hive Sync");
if (syncTasks == null) {
syncTasks = new ScheduledTasks<>(ThreadPool.Type.HIVE_SYNC);
}
List<TableMetadata> tables = ServiceContainer.getMetaService().listTables();
Set<TableIdentifier> ids =
tables.stream().map(TableMetadata::getTableIdentifier).collect(Collectors.toSet());
syncTasks.checkRunningTask(ids,
() -> 0L,
() -> SYNC_INTERVAL,
SupportHiveSyncService.SupportHiveSyncTask::new,
false);
LOG.info("Schedule Support Hive Sync finished with {} valid ids", ids.size());
} catch (Throwable t) {
LOG.error("Schedule Support Hive Sync failed", t);
}
List<TableMetadata> tables = ServiceContainer.getMetaService().listTables();
Set<TableIdentifier> ids =
tables.stream().map(TableMetadata::getTableIdentifier).collect(Collectors.toSet());
syncTasks.checkRunningTask(ids,
() -> 0L,
() -> SYNC_INTERVAL,
SupportHiveSyncService.SupportHiveSyncTask::new,
false);
LOG.info("Schedule Support Hive Sync finished with {} valid ids", ids.size());
}

@Override
Expand Down
28 changes: 16 additions & 12 deletions ...s-server/src/main/java/com/netease/arctic/ams/server/service/impl/TableExpireService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,19 +74,23 @@ public class TableExpireService implements ITableExpireService {

@Override
public synchronized void checkTableExpireTasks() {
LOG.info("Schedule Expired Cleaner");
if (cleanTasks == null) {
cleanTasks = new ScheduledTasks<>(ThreadPool.Type.EXPIRE);
try {
LOG.info("Schedule Expired Cleaner");
if (cleanTasks == null) {
cleanTasks = new ScheduledTasks<>(ThreadPool.Type.EXPIRE);
}
List<TableMetadata> tables = ServiceContainer.getMetaService().listTables();
Set<TableIdentifier> ids =
tables.stream().map(TableMetadata::getTableIdentifier).collect(Collectors.toSet());
cleanTasks.checkRunningTask(ids,
() -> 0L,
() -> EXPIRE_INTERVAL,
TableExpireTask::new,
false);
LOG.info("Schedule Expired Cleaner finished with {} valid ids", ids.size());
} catch (Throwable t) {
LOG.error("unexpected error when checkTableExpireTasks", t);
}
List<TableMetadata> tables = ServiceContainer.getMetaService().listTables();
Set<TableIdentifier> ids =
tables.stream().map(TableMetadata::getTableIdentifier).collect(Collectors.toSet());
cleanTasks.checkRunningTask(ids,
() -> 0L,
() -> EXPIRE_INTERVAL,
TableExpireTask::new,
false);
LOG.info("Schedule Expired Cleaner finished with {} valid ids", ids.size());
}

@Override
Expand Down
30 changes: 17 additions & 13 deletions ...ms-server/src/main/java/com/netease/arctic/ams/server/service/impl/TrashCleanService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -53,20 +53,24 @@ public class TrashCleanService implements Closeable {
private ScheduledTasks<TableIdentifier, TableTrashCleanTask> cleanTasks;

public synchronized void checkTrashCleanTasks() {
LOG.info("Schedule Trash Cleaner");
if (cleanTasks == null) {
cleanTasks = new ScheduledTasks<>(ThreadPool.Type.TRASH_CLEAN);
}
try {
LOG.info("Schedule Trash Cleaner");
if (cleanTasks == null) {
cleanTasks = new ScheduledTasks<>(ThreadPool.Type.TRASH_CLEAN);
}

List<TableMetadata> tables = ServiceContainer.getMetaService().listTables();
Set<TableIdentifier> ids =
tables.stream().map(TableMetadata::getTableIdentifier).collect(Collectors.toSet());
cleanTasks.checkRunningTask(ids,
() -> RandomUtils.nextLong(0, CHECK_INTERVAL),
() -> CHECK_INTERVAL,
TableTrashCleanTask::new,
true);
LOG.info("Schedule Trash Cleaner finished with {} tasks", ids.size());
List<TableMetadata> tables = ServiceContainer.getMetaService().listTables();
Set<TableIdentifier> ids =
tables.stream().map(TableMetadata::getTableIdentifier).collect(Collectors.toSet());
cleanTasks.checkRunningTask(ids,
() -> RandomUtils.nextLong(0, CHECK_INTERVAL),
() -> CHECK_INTERVAL,
TableTrashCleanTask::new,
true);
LOG.info("Schedule Trash Cleaner finished with {} tasks", ids.size());
} catch (Throwable t) {
LOG.error("Schedule Trash Cleaner unexpected error", t);
}
}

@Override
Expand Down