You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampark.apache.org by be...@apache.org on 2023/01/11 11:07:23 UTC
[incubator-streampark] branch dev updated: [Improve] read flink on k8s deploy log improvement (#2243)
This is an automated email from the ASF dual-hosted git repository.
benjobs pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git
The following commit(s) were added to refs/heads/dev by this push:
new f8fe8406b [Improve] read flink on k8s deploy log improvement (#2243)
f8fe8406b is described below
commit f8fe8406bfc8c31942af4d3c6093c68d3e31ec77
Author: benjobs <be...@apache.org>
AuthorDate: Wed Jan 11 19:07:16 2023 +0800
[Improve] read flink on k8s deploy log improvement (#2243)
* [Improve] CompletableFuture improvement
* [Improve] read flink on k8s deploy log improvement
* [Improve] call CompletableFuture method change to CompletableFutureUtils
* [Improve] CompletableFutureUtils improvement
* [Improve] exception testcase improvement
---
.../common/util/CompletableFutureUtils.scala | 45 ++++----
.../streampark/console/base/util/MoreFutures.java | 125 ---------------------
.../core/controller/ApplicationController.java | 25 +----
.../console/core/service/ApplicationService.java | 2 +
.../console/core/service/LogClientService.java | 48 +++-----
.../console/core/service/LoggerService.java | 25 -----
.../core/service/impl/ApplicationServiceImpl.java | 42 ++++++-
.../core/service/impl/LoggerServiceImpl.java | 65 -----------
.../common/util/CompletableFutureUtilsTest.java} | 81 ++++++++++++-
.../src/api/flink/app/app.ts | 6 +-
.../flink/app/components/AppView/LogModal.vue | 20 ++--
11 files changed, 179 insertions(+), 305 deletions(-)
diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/util/CompletableFutureUtils.scala b/streampark-common/src/main/scala/org/apache/streampark/common/util/CompletableFutureUtils.scala
index cc5bfdeb9..838c4f6c7 100644
--- a/streampark-common/src/main/scala/org/apache/streampark/common/util/CompletableFutureUtils.scala
+++ b/streampark-common/src/main/scala/org/apache/streampark/common/util/CompletableFutureUtils.scala
@@ -17,8 +17,8 @@
package org.apache.streampark.common.util
-import java.util.concurrent.{Callable, CompletableFuture, ScheduledThreadPoolExecutor, ThreadFactory, TimeoutException, TimeUnit}
-import java.util.function.{Consumer, Function => JavaFunc}
+import java.util.concurrent.{Callable, CompletableFuture, ScheduledThreadPoolExecutor, ThreadFactory, TimeUnit, TimeoutException}
+import java.util.function.{BiConsumer, Consumer, Function => JavaFunc}
object CompletableFutureUtils {
@@ -39,7 +39,7 @@ object CompletableFutureUtils {
delayer
}
- def setTimeout[T](timeout: Long, unit: TimeUnit): CompletableFuture[T] = {
+ private[this] def setTimeout[T](timeout: Long, unit: TimeUnit): CompletableFuture[T] = {
val result = new CompletableFuture[T]
completableDelayer.schedule(
new Callable[Boolean] {
@@ -56,7 +56,16 @@ object CompletableFutureUtils {
unit: TimeUnit,
handle: JavaFunc[T, T],
exceptionally: JavaFunc[Throwable, T]): CompletableFuture[T] = {
- future.applyToEither(setTimeout(timeout, unit), handle).exceptionally(exceptionally)
+ future
+ .applyToEither(setTimeout(timeout, unit), handle)
+ .exceptionally(exceptionally)
+ .whenComplete(new BiConsumer[T, Throwable]() {
+ override def accept(t: T, u: Throwable): Unit = {
+ if (!future.isDone) {
+ future.cancel(true)
+ }
+ }
+ })
}
/**
@@ -75,36 +84,34 @@ object CompletableFutureUtils {
unit: TimeUnit,
handle: Consumer[T],
exceptionally: Consumer[Throwable]): CompletableFuture[Unit] = {
+
future.applyToEither(
setTimeout(timeout, unit),
- new JavaFunc[T, Unit] {
+ new JavaFunc[T, Unit]() {
override def apply(t: T): Unit = {
if (handle != null) {
handle.accept(t)
}
}
- }).exceptionally(new JavaFunc[Throwable, Unit] {
+ }).exceptionally(new JavaFunc[Throwable, Unit]() {
override def apply(t: Throwable): Unit = {
if (exceptionally != null) {
exceptionally.accept(t)
}
}
+ }).whenComplete(new BiConsumer[Unit, Throwable]() {
+ override def accept(t: Unit, u: Throwable): Unit = {
+ if (!future.isDone) {
+ future.cancel(true)
+ }
+ }
})
}
- def runTimeout[T](future: CompletableFuture[T], timeout: Long, unit: TimeUnit): CompletableFuture[Unit] = {
- runTimeout(
- future,
- timeout,
- unit,
- null,
- new Consumer[Throwable] {
- override def accept(t: Throwable): Unit = {
- if (!future.isDone) {
- future.cancel(true)
- }
- }
- })
+ def runTimeout[T](future: CompletableFuture[T],
+ timeout: Long,
+ unit: TimeUnit): CompletableFuture[Unit] = {
+ runTimeout(future, timeout, unit, null, null)
}
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/util/MoreFutures.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/util/MoreFutures.java
deleted file mode 100644
index 78fbecf7b..000000000
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/util/MoreFutures.java
+++ /dev/null
@@ -1,125 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.streampark.console.base.util;
-
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionStage;
-import java.util.concurrent.Delayed;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.function.Function;
-import java.util.function.Supplier;
-
-public final class MoreFutures {
- private MoreFutures() {}
-
- public static <T> T await(CompletionStage<T> stage) {
- CompletableFuture<T> future = stage.toCompletableFuture();
-
- try {
- return future.get();
- } catch (InterruptedException e) {
- throw new RuntimeException(e);
- } catch (ExecutionException e) {
- throw new RuntimeException(e.getCause());
- }
- }
-
- public static <T> T deref(CompletionStage<T> stage, long timeout, TimeUnit unit) {
- CompletableFuture<T> future = stage.toCompletableFuture();
-
- try {
- return future.get(timeout, unit);
- } catch (TimeoutException | InterruptedException e) {
- throw new RuntimeException(e);
- } catch (ExecutionException e) {
- throw new RuntimeException(e.getCause());
- }
- }
-
- public static <T> T derefUsingDefaultTimeout(CompletionStage<T> stage) {
- return deref(stage, 5L, TimeUnit.SECONDS);
- }
-
- public static <T> CompletionStage<T> exceptionallyCompose(
- CompletionStage<T> stage, Function<Throwable, ? extends CompletionStage<T>> fn) {
- return stage
- .thenApply(CompletableFuture::completedFuture)
- .exceptionally((Function<Throwable, ? extends CompletableFuture<T>>) fn)
- .thenCompose(Function.identity());
- }
-
- public static <T> CompletableFuture<T> completeImmediately(Supplier<T> supplier) {
- try {
- return CompletableFuture.completedFuture(supplier.get());
- } catch (Throwable e) {
- CompletableFuture<T> f = new CompletableFuture<>();
- f.completeExceptionally(e);
- return f;
- }
- }
-
- public static <T> CompletableFuture<T> completedExceptionally(Throwable cause) {
- CompletableFuture<T> future = new CompletableFuture<>();
- future.completeExceptionally(cause);
- return future;
- }
-
- public static <T> ScheduledFuture<T> completedScheduledFuture(T value) {
- return new CompletedScheduledFuture<>(value);
- }
-
- private static final class CompletedScheduledFuture<T> implements ScheduledFuture<T> {
- private final T value;
-
- private CompletedScheduledFuture(T value) {
- this.value = value;
- }
-
- public long getDelay(TimeUnit unit) {
- return 0L;
- }
-
- public int compareTo(Delayed o) {
- return Long.compare(0L, o.getDelay(TimeUnit.MILLISECONDS));
- }
-
- public boolean cancel(boolean mayInterruptIfRunning) {
- return false;
- }
-
- public boolean isCancelled() {
- return false;
- }
-
- public boolean isDone() {
- return true;
- }
-
- public T get() throws InterruptedException, ExecutionException {
- return this.value;
- }
-
- public T get(long timeout, TimeUnit unit)
- throws InterruptedException, ExecutionException, TimeoutException {
- return this.value;
- }
- }
-}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationController.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationController.java
index 1683e208a..a580d7655 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationController.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ApplicationController.java
@@ -23,7 +23,6 @@ import org.apache.streampark.console.base.domain.ApiDocConstant;
import org.apache.streampark.console.base.domain.RestRequest;
import org.apache.streampark.console.base.domain.RestResponse;
import org.apache.streampark.console.base.exception.InternalException;
-import org.apache.streampark.console.base.util.MoreFutures;
import org.apache.streampark.console.core.annotation.ApiAccess;
import org.apache.streampark.console.core.bean.AppControl;
import org.apache.streampark.console.core.entity.Application;
@@ -34,7 +33,6 @@ import org.apache.streampark.console.core.service.AppBuildPipeService;
import org.apache.streampark.console.core.service.ApplicationBackUpService;
import org.apache.streampark.console.core.service.ApplicationLogService;
import org.apache.streampark.console.core.service.ApplicationService;
-import org.apache.streampark.console.core.service.LoggerService;
import org.apache.streampark.flink.packer.pipeline.PipelineStatus;
import com.baomidou.mybatisplus.core.metadata.IPage;
@@ -48,7 +46,6 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.multipart.MultipartFile;
import springfox.documentation.annotations.ApiIgnore;
@@ -77,8 +74,6 @@ public class ApplicationController {
@Autowired private AppBuildPipeService appBuildPipeService;
- @Autowired private LoggerService logService;
-
@ApiAccess
@PostMapping("get")
@RequiresPermissions("app:detail")
@@ -417,21 +412,13 @@ public class ApplicationController {
@ApiOperation(value = "Read flink on k8s deploy log")
@ApiImplicitParams({
- @ApiImplicitParam(name = "namespace", value = "ks8 namespace"),
- @ApiImplicitParam(name = "jobName", value = "job name"),
- @ApiImplicitParam(name = "jobId", value = "job id"),
- @ApiImplicitParam(name = "skipLineNum", value = "number of log lines skipped loading"),
+ @ApiImplicitParam(name = "id", value = "app id"),
+ @ApiImplicitParam(name = "offset", value = "number of log lines offset"),
@ApiImplicitParam(name = "limit", value = "number of log lines loaded at once")
})
- @PostMapping(value = "/detail")
- public RestResponse detail(
- @RequestParam(value = "namespace", required = false) String namespace,
- @RequestParam(value = "jobName", required = false) String jobName,
- @RequestParam(value = "jobId", required = false) String jobId,
- @RequestParam(value = "skipLineNum", required = false) Integer skipLineNum,
- @RequestParam(value = "limit", required = false) Integer limit) {
- return RestResponse.success(
- MoreFutures.derefUsingDefaultTimeout(
- logService.queryLog(namespace, jobName, jobId, skipLineNum, limit)));
+ @PostMapping(value = "k8sStartLog")
+ public RestResponse k8sStartLog(Long id, Integer offset, Integer limit) throws Exception {
+ String resp = applicationService.k8sStartLog(id, offset, limit);
+ return RestResponse.success(resp);
}
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/ApplicationService.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/ApplicationService.java
index 1ab095855..87513f09b 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/ApplicationService.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/ApplicationService.java
@@ -113,4 +113,6 @@ public interface ApplicationService extends IService<Application> {
List<String> getRecentK8sTmPodTemplate();
List<String> historyUploadJars();
+
+ String k8sStartLog(Long id, Integer offset, Integer limit) throws Exception;
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/LogClientService.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/LogClientService.java
index 9b065b6c1..c9525ab64 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/LogClientService.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/LogClientService.java
@@ -17,14 +17,15 @@
package org.apache.streampark.console.core.service;
+import org.apache.streampark.console.base.exception.ApiDetailException;
+
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import java.io.File;
-import java.io.IOException;
+import java.io.FileNotFoundException;
import java.nio.file.Files;
import java.nio.file.Paths;
-import java.util.Collections;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;
@@ -33,38 +34,21 @@ import java.util.stream.Stream;
@Slf4j
@Component
public class LogClientService {
- public String rollViewLog(String path, int skipLineNum, int limit) {
- String result = "";
+ public String rollViewLog(String path, int offset, int limit) {
try {
- List<String> lines = readPartFileContent(path, skipLineNum, limit);
- StringBuilder builder = new StringBuilder();
- lines.forEach(line -> builder.append(line).append("\r\n"));
- return builder.toString();
- } catch (Exception e) {
- log.error("roll view log error", e);
- }
- return result;
- }
-
- /**
- * read part file content，can skip any line and read some lines
- *
- * @param filePath file path
- * @param skipLine skip line
- * @param limit read lines limit
- * @return part file content
- */
- private List<String> readPartFileContent(String filePath, int skipLine, int limit) {
- File file = new File(filePath);
- if (file.exists() && file.isFile()) {
- try (Stream<String> stream = Files.lines(Paths.get(filePath))) {
- return stream.skip(skipLine).limit(limit).collect(Collectors.toList());
- } catch (IOException e) {
- log.error("read file error", e);
+ File file = new File(path);
+ if (file.exists() && file.isFile()) {
+ try (Stream<String> stream = Files.lines(Paths.get(path))) {
+ List<String> lines = stream.skip(offset).limit(limit).collect(Collectors.toList());
+ StringBuilder builder = new StringBuilder();
+ lines.forEach(line -> builder.append(line).append("\r\n"));
+ return builder.toString();
+ }
+ } else {
+ throw new FileNotFoundException("file path: " + path + " not exists ");
}
- } else {
- log.info("file path: {} not exists", filePath);
+ } catch (Exception e) {
+ throw new ApiDetailException("roll view log error: " + e);
}
- return Collections.emptyList();
}
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/LoggerService.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/LoggerService.java
deleted file mode 100644
index 5e0df1ca4..000000000
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/LoggerService.java
+++ /dev/null
@@ -1,25 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.streampark.console.core.service;
-
-import java.util.concurrent.CompletionStage;
-
-public interface LoggerService {
- CompletionStage<String> queryLog(
- String namespac, String jobName, String jobId, int skipLineNum, int limit);
-}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
index f10622d51..778dcfe67 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
@@ -72,6 +72,7 @@ import org.apache.streampark.console.core.service.EffectiveService;
import org.apache.streampark.console.core.service.FlinkClusterService;
import org.apache.streampark.console.core.service.FlinkEnvService;
import org.apache.streampark.console.core.service.FlinkSqlService;
+import org.apache.streampark.console.core.service.LogClientService;
import org.apache.streampark.console.core.service.ProjectService;
import org.apache.streampark.console.core.service.SavePointService;
import org.apache.streampark.console.core.service.SettingService;
@@ -206,6 +207,8 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
@Autowired private VariableService variableService;
+ @Autowired private LogClientService logClient;
+
@PostConstruct
public void resetOptionState() {
this.baseMapper.resetOptionState();
@@ -579,6 +582,43 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
.collect(Collectors.toList());
}
+ @Override
+ public String k8sStartLog(Long id, Integer offset, Integer limit) throws Exception {
+ Application application = getById(id);
+ AssertUtils.state(application != null);
+ if (ExecutionMode.isKubernetesMode(application.getExecutionModeEnum())) {
+ CompletableFuture<String> future =
+ CompletableFuture.supplyAsync(
+ () ->
+ KubernetesDeploymentHelper.watchDeploymentLog(
+ application.getK8sNamespace(),
+ application.getJobName(),
+ application.getJobId()));
+
+ return CompletableFutureUtils.supplyTimeout(
+ future,
+ 5,
+ TimeUnit.SECONDS,
+ success -> success,
+ exception -> {
+ String errorLog =
+ String.format(
+ "%s/%s_err.log", WebUtils.getAppTempDir(), application.getJobId());
+ File file = new File(errorLog);
+ if (file.exists()) {
+ return errorLog;
+ } else {
+ throw new ApiDetailException("get k8s job log failed: " + exception.getMessage());
+ }
+ })
+ .thenApply(path -> logClient.rollViewLog(path, offset, limit))
+ .get();
+ } else {
+ throw new ApiAlertException(
+ "job executionMode must be kubernetes-session|kubernetes-application");
+ }
+ }
+
@Override
public String getYarnName(Application appParam) {
String[] args = new String[2];
@@ -1254,7 +1294,6 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
IngressController.deleteIngress(
application.getK8sNamespace(), application.getJobName());
}
- cancelFuture.cancel(true);
cancelFutureMap.remove(application.getId());
});
}
@@ -1558,7 +1597,6 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
})
.whenComplete(
(t, e) -> {
- future.cancel(true);
startFutureMap.remove(application.getId());
});
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/LoggerServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/LoggerServiceImpl.java
deleted file mode 100644
index ba1b8dd5d..000000000
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/LoggerServiceImpl.java
+++ /dev/null
@@ -1,65 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.streampark.console.core.service.impl;
-
-import org.apache.streampark.console.base.util.WebUtils;
-import org.apache.streampark.console.core.service.LogClientService;
-import org.apache.streampark.console.core.service.LoggerService;
-import org.apache.streampark.flink.kubernetes.helper.KubernetesDeploymentHelper;
-
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
-
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CompletionStage;
-
-/** log service */
-@Service
-@Slf4j
-public class LoggerServiceImpl implements LoggerService {
-
- @Autowired private LogClientService logClient;
-
- /**
- * view log
- *
- * @param skipLineNum skip line number
- * @param limit limit
- * @return log string data
- */
- @Override
- public CompletionStage<String> queryLog(
- String nameSpace, String jobName, String jobId, int skipLineNum, int limit) {
- return CompletableFuture.supplyAsync(() -> jobDeploymentsWatch(nameSpace, jobName, jobId))
- .exceptionally(
- e -> {
- try {
- return String.format("%s/%s_err.log", WebUtils.getAppTempDir(), jobId);
- } catch (Exception ex) {
- log.error("Generate log path exception:{}", ex.getMessage());
- return null;
- }
- })
- .thenApply(path -> logClient.rollViewLog(String.valueOf(path), skipLineNum, limit));
- }
-
- private String jobDeploymentsWatch(String nameSpace, String jobName, String jobId) {
- return KubernetesDeploymentHelper.watchDeploymentLog(nameSpace, jobName, jobId);
- }
-}
diff --git a/streampark-console/streampark-console-service/src/test/java/java/util/concurrent/CompletableFutureTest.java b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/common/util/CompletableFutureUtilsTest.java
similarity index 66%
rename from streampark-console/streampark-console-service/src/test/java/java/util/concurrent/CompletableFutureTest.java
rename to streampark-console/streampark-console-service/src/test/java/org/apache/streampark/common/util/CompletableFutureUtilsTest.java
index 99838b55f..d6c226b74 100644
--- a/streampark-console/streampark-console-service/src/test/java/java/util/concurrent/CompletableFutureTest.java
+++ b/streampark-console/streampark-console-service/src/test/java/org/apache/streampark/common/util/CompletableFutureUtilsTest.java
@@ -15,16 +15,20 @@
* limitations under the License.
*/
-package java.util.concurrent;
-
-import org.apache.streampark.common.util.CompletableFutureUtils;
+package org.apache.streampark.common.util;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
-class CompletableFutureTest {
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class CompletableFutureUtilsTest {
@Test
void testStartJobNormally() throws Exception {
@@ -143,4 +147,73 @@ class CompletableFutureTest {
}
return "start successful";
}
+
+ @Test
+ public void thenFutureNormally() throws Exception {
+ String successResult = "success";
+ String exceptionResult = "error";
+
+ String resp =
+ CompletableFutureUtils.supplyTimeout(
+ CompletableFuture.supplyAsync(() -> successResult),
+ 1,
+ TimeUnit.SECONDS,
+ success -> success,
+ e -> exceptionResult)
+ .thenApply(r -> r)
+ .get();
+
+ Assertions.assertEquals(resp, successResult);
+ }
+
+ @Test
+ public void thenFutureTimeout() throws Exception {
+ String successResult = "success";
+ String exceptionResult = "error";
+ CompletableFuture<String> future =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ Thread.sleep(5000);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ return successResult;
+ });
+ String resp =
+ CompletableFutureUtils.supplyTimeout(
+ future, 1, TimeUnit.SECONDS, success -> success, e -> exceptionResult)
+ .thenApply(r -> r)
+ .get();
+ Assertions.assertEquals(resp, exceptionResult);
+ }
+
+ @Test
+ public void thenFutureException() {
+ String expectedExceptionMessage = "Exception in exceptionally handler";
+ int processTime = 5000;
+ int timeOut = 1000;
+ CompletableFuture<String> future =
+ CompletableFutureUtils.supplyTimeout(
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ Thread.sleep(processTime);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ return "success";
+ }),
+ timeOut,
+ TimeUnit.MILLISECONDS,
+ success -> success,
+ e -> {
+ throw new RuntimeException(expectedExceptionMessage);
+ });
+
+ assertThatThrownBy(future::get)
+ .getCause()
+ .as(expectedExceptionMessage)
+ .isInstanceOf(RuntimeException.class);
+ }
}
diff --git a/streampark-console/streampark-console-webapp/src/api/flink/app/app.ts b/streampark-console/streampark-console-webapp/src/api/flink/app/app.ts
index 958c2eb7a..054b22f33 100644
--- a/streampark-console/streampark-console-webapp/src/api/flink/app/app.ts
+++ b/streampark-console/streampark-console-webapp/src/api/flink/app/app.ts
@@ -25,7 +25,7 @@ enum APP_API {
UPDATE = '/flink/app/update',
COPY = '/flink/app/copy',
UPLOAD = '/flink/app/upload',
- START_LOG = '/flink/app/detail',
+ K8S_START_LOG = '/flink/app/k8sStartLog',
DEPLOY = '/flink/app/deploy',
MAPPING = '/flink/app/mapping',
YARN = '/flink/app/yarn',
@@ -188,8 +188,8 @@ export function fetchMapping(data): Promise<boolean> {
* @param params
* @returns {Promise<AxiosResponse<any>>}
*/
-export function fetchStartLog(data): Promise<AxiosResponse<any>> {
- return defHttp.post({ url: APP_API.START_LOG, data }, { isReturnNativeResponse: true });
+export function fetchK8sStartLog(data): Promise<AxiosResponse<any>> {
+ return defHttp.post({ url: APP_API.K8S_START_LOG, data }, { isReturnNativeResponse: true });
}
/**
* SavepointPath
diff --git a/streampark-console/streampark-console-webapp/src/views/flink/app/components/AppView/LogModal.vue b/streampark-console/streampark-console-webapp/src/views/flink/app/components/AppView/LogModal.vue
index 20ab2d608..29cd70417 100644
--- a/streampark-console/streampark-console-webapp/src/views/flink/app/components/AppView/LogModal.vue
+++ b/streampark-console/streampark-console-webapp/src/views/flink/app/components/AppView/LogModal.vue
@@ -24,7 +24,7 @@
</script>
<script setup lang="ts" name="LogModal">
import { reactive, ref, unref } from 'vue';
- import { fetchStartLog } from '/@/api/flink/app/app';
+ import { fetchK8sStartLog } from '/@/api/flink/app/app';
import { BasicModal, useModalInner } from '/@/components/Modal';
import { Icon } from '/@/components/Icon';
import { formatToDateTime } from '/@/utils/dateUtil';
@@ -34,7 +34,7 @@
const logTime = ref<string>('');
const getLogLoading = ref<boolean>(false);
const app = reactive<Recordable>({});
- let skipLineNum = 0;
+ let offset = 0;
let logContent = '';
const [registerModal, { changeLoading, closeModal }] = useModalInner((data) => {
@@ -44,7 +44,7 @@
const { setContent, logRef, handleRevealLine, getLineCount } = useLog();
function onReceiveModalData(data) {
- skipLineNum = 0;
+ offset = 0;
logContent = '';
Object.assign(app, unref(data.app));
changeLoading(true);
@@ -63,12 +63,10 @@
async function refreshLog() {
try {
getLogLoading.value = true;
- const { data } = await fetchStartLog({
- namespace: app.k8sNamespace,
- jobName: app.jobName,
- jobId: app.jobId,
+ const { data } = await fetchK8sStartLog({
+ id: app.id,
+ offset: offset,
limit: 100,
- skipLineNum,
});
const status = data.status || 'error';
if (status === 'success') {
@@ -78,10 +76,10 @@
handleRevealLine();
setTimeout(() => {
const currentLineCount = getLineCount() - 1;
- if (currentLineCount < skipLineNum + 100) {
- skipLineNum = currentLineCount;
+ if (currentLineCount < offset + 100) {
+ offset = currentLineCount;
} else {
- skipLineNum += 100;
+ offset += 100;
}
}, 500);
}