lanyuanxiaoyao/hudi-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

perf(web): 增强查询同步压缩任务存在的性能

Browse Source 
先判断是否有同步压缩锁，有再获取运行指标
This commit is contained in:
兰缘小妖
2023-05-14 00:50:54 +08:00
parent 41ca5319d2
commit 3f0d9f11da
1 changed files with 36 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/TableController.java
View File

@@ -92,8 +92,10 @@ public class TableController extends BaseController {
CompletableFuture<FlinkJob> flinkJobFuture = CompletableFuture.supplyAsync(() -> infoService.flinkJobDetail(item.getId()), EXECUTOR); CompletableFuture<FlinkJob> flinkJobFuture = CompletableFuture.supplyAsync(() -> infoService.flinkJobDetail(item.getId()), EXECUTOR);
CompletableFuture<TableMeta> tableMetaFuture = CompletableFuture.supplyAsync(() -> infoService.tableMetaDetail(item.getId(), item.getAlias()), EXECUTOR); CompletableFuture<TableMeta> tableMetaFuture = CompletableFuture.supplyAsync(() -> infoService.tableMetaDetail(item.getId(), item.getAlias()), EXECUTOR);
CompletableFuture<SyncState> syncStateFuture = CompletableFuture.supplyAsync(() -> infoService.syncStateDetail(item.getId(), item.getAlias()), EXECUTOR); CompletableFuture<SyncState> syncStateFuture = CompletableFuture.supplyAsync(() -> infoService.syncStateDetail(item.getId(), item.getAlias()), EXECUTOR);
CompletableFuture<Boolean> syncRunning = CompletableFuture.supplyAsync(() -> zookeeperService.existsPath(NameHelper.syncRunningLockPath(item.getId())), EXECUTOR); CompletableFuture<RunMeta> syncRuntime = CompletableFuture
CompletableFuture<RunMeta> syncRuntime = CompletableFuture.supplyAsync(() -> { .supplyAsync(() -> zookeeperService.existsPath(NameHelper.syncRunningLockPath(item.getId())), EXECUTOR)
.thenApply(running -> {
if (running) {
try { try {
String data = zookeeperService.getData(NameHelper.syncRunningLockPath(item.getId())); String data = zookeeperService.getData(NameHelper.syncRunningLockPath(item.getId()));
if (StrUtil.isNotBlank(data)) { if (StrUtil.isNotBlank(data)) {
@@ -102,38 +104,42 @@ public class TableController extends BaseController {
} catch (Exception e) { } catch (Exception e) {
logger.error("Parse sync runtime info for: {}", item.getId()); logger.error("Parse sync runtime info for: {}", item.getId());
} }
}
return null; return null;
}, EXECUTOR); });
CompletableFuture<Boolean> compactionRunning = CompletableFuture.supplyAsync(() -> zookeeperService.existsPath(NameHelper.compactionRunningLockPath(item.getId(), item.getAlias())), EXECUTOR); CompletableFuture<RunMeta> compactionRuntime = CompletableFuture
CompletableFuture<RunMeta> compactionRuntime = CompletableFuture.supplyAsync(() -> { .supplyAsync(() -> zookeeperService.existsPath(NameHelper.compactionRunningLockPath(item.getId(), item.getAlias())), EXECUTOR)
.thenApply(running -> {
if (running) {
try { try {
String data = zookeeperService.getData(NameHelper.compactionRunningLockPath(item.getId(), item.getAlias())); String data = zookeeperService.getData(NameHelper.compactionRunningLockPath(item.getId(), item.getAlias()));
if (StrUtil.isNotBlank(data)) { if (StrUtil.isNotBlank(data)) {
return mapper.readValue(data, RunMeta.class); return mapper.readValue(data, RunMeta.class);
} }
} catch (Exception e) { } catch (Exception e) {
logger.error("Parse compaction runtime info for: {}", item.getId()); logger.error("Parse sync runtime info for: {}", item.getId());
}
} }
return null; return null;
}, EXECUTOR); });
try { try {
CompletableFuture.allOf( CompletableFuture.allOf(
flinkJobFuture, flinkJobFuture,
tableMetaFuture, tableMetaFuture,
syncStateFuture, syncStateFuture,
syncRunning,
syncRuntime, syncRuntime,
compactionRunning,
compactionRuntime compactionRuntime
).get(); ).get();
RunMeta syncRunMeta = syncRuntime.get();
RunMeta compactionRunMeta = compactionRuntime.get();
return new TableVO( return new TableVO(
flinkJobFuture.get(), flinkJobFuture.get(),
tableMetaFuture.get(), tableMetaFuture.get(),
new SyncStateVO(syncStateFuture.get()), new SyncStateVO(syncStateFuture.get()),
syncRunning.get(), ObjectUtil.isNotNull(syncRunMeta),
syncRuntime.get(), syncRunMeta,
compactionRunning.get(), ObjectUtil.isNotNull(compactionRunMeta),
compactionRuntime.get() compactionRunMeta
); );
} catch (InterruptedException | ExecutionException e) { } catch (InterruptedException | ExecutionException e) {
logger.error("Something bad", e); logger.error("Something bad", e);