diff --git a/service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/TableController.java b/service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/TableController.java index 57df1c4..88bec54 100644 --- a/service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/TableController.java +++ b/service-web/src/main/java/com/lanyuanxiaoyao/service/web/controller/TableController.java @@ -92,48 +92,54 @@ public class TableController extends BaseController { CompletableFuture flinkJobFuture = CompletableFuture.supplyAsync(() -> infoService.flinkJobDetail(item.getId()), EXECUTOR); CompletableFuture tableMetaFuture = CompletableFuture.supplyAsync(() -> infoService.tableMetaDetail(item.getId(), item.getAlias()), EXECUTOR); CompletableFuture syncStateFuture = CompletableFuture.supplyAsync(() -> infoService.syncStateDetail(item.getId(), item.getAlias()), EXECUTOR); - CompletableFuture syncRunning = CompletableFuture.supplyAsync(() -> zookeeperService.existsPath(NameHelper.syncRunningLockPath(item.getId())), EXECUTOR); - CompletableFuture syncRuntime = CompletableFuture.supplyAsync(() -> { - try { - String data = zookeeperService.getData(NameHelper.syncRunningLockPath(item.getId())); - if (StrUtil.isNotBlank(data)) { - return mapper.readValue(data, RunMeta.class); - } - } catch (Exception e) { - logger.error("Parse sync runtime info for: {}", item.getId()); - } - return null; - }, EXECUTOR); - CompletableFuture compactionRunning = CompletableFuture.supplyAsync(() -> zookeeperService.existsPath(NameHelper.compactionRunningLockPath(item.getId(), item.getAlias())), EXECUTOR); - CompletableFuture compactionRuntime = CompletableFuture.supplyAsync(() -> { - try { - String data = zookeeperService.getData(NameHelper.compactionRunningLockPath(item.getId(), item.getAlias())); - if (StrUtil.isNotBlank(data)) { - return mapper.readValue(data, RunMeta.class); - } - } catch (Exception e) { - logger.error("Parse compaction runtime info for: {}", item.getId()); - } - return null; - }, EXECUTOR); + CompletableFuture syncRuntime = CompletableFuture + .supplyAsync(() -> zookeeperService.existsPath(NameHelper.syncRunningLockPath(item.getId())), EXECUTOR) + .thenApply(running -> { + if (running) { + try { + String data = zookeeperService.getData(NameHelper.syncRunningLockPath(item.getId())); + if (StrUtil.isNotBlank(data)) { + return mapper.readValue(data, RunMeta.class); + } + } catch (Exception e) { + logger.error("Parse sync runtime info for: {}", item.getId()); + } + } + return null; + }); + CompletableFuture compactionRuntime = CompletableFuture + .supplyAsync(() -> zookeeperService.existsPath(NameHelper.compactionRunningLockPath(item.getId(), item.getAlias())), EXECUTOR) + .thenApply(running -> { + if (running) { + try { + String data = zookeeperService.getData(NameHelper.compactionRunningLockPath(item.getId(), item.getAlias())); + if (StrUtil.isNotBlank(data)) { + return mapper.readValue(data, RunMeta.class); + } + } catch (Exception e) { + logger.error("Parse sync runtime info for: {}", item.getId()); + } + } + return null; + }); try { CompletableFuture.allOf( flinkJobFuture, tableMetaFuture, syncStateFuture, - syncRunning, syncRuntime, - compactionRunning, compactionRuntime ).get(); + RunMeta syncRunMeta = syncRuntime.get(); + RunMeta compactionRunMeta = compactionRuntime.get(); return new TableVO( flinkJobFuture.get(), tableMetaFuture.get(), new SyncStateVO(syncStateFuture.get()), - syncRunning.get(), - syncRuntime.get(), - compactionRunning.get(), - compactionRuntime.get() + ObjectUtil.isNotNull(syncRunMeta), + syncRunMeta, + ObjectUtil.isNotNull(compactionRunMeta), + compactionRunMeta ); } catch (InterruptedException | ExecutionException e) { logger.error("Something bad", e);