You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gy...@apache.org on 2022/06/30 09:05:19 UTC
[flink-kubernetes-operator] branch main updated: [FLINK-27868] Harden running job check
This is an automated email from the ASF dual-hosted git repository.
gyfora pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-kubernetes-operator.git
The following commit(s) were added to refs/heads/main by this push:
new 015b929 [FLINK-27868] Harden running job check
015b929 is described below
commit 015b929912479458adfad52e378778f4e6f14163
Author: Matyas Orhidi <ma...@apple.com>
AuthorDate: Wed Jun 29 08:16:17 2022 +0200
[FLINK-27868] Harden running job check
---
.../kubernetes/operator/service/FlinkService.java | 43 +++++++++++++--
.../kubernetes/operator/TestingFlinkService.java | 33 ++++++++++++
.../operator/service/FlinkServiceTest.java | 62 ++++++++++++++++++++++
3 files changed, 135 insertions(+), 3 deletions(-)
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java
index e2d2994..aca8cff 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/FlinkService.java
@@ -52,9 +52,12 @@ import org.apache.flink.kubernetes.operator.observer.SavepointFetchResult;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
import org.apache.flink.runtime.client.JobStatusMessage;
+import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.highavailability.nonha.standalone.StandaloneClientHAServices;
import org.apache.flink.runtime.jobgraph.RestoreMode;
import org.apache.flink.runtime.jobmaster.JobResult;
+import org.apache.flink.runtime.messages.webmonitor.JobDetails;
+import org.apache.flink.runtime.messages.webmonitor.MultipleJobsDetails;
import org.apache.flink.runtime.rest.FileUpload;
import org.apache.flink.runtime.rest.RestClient;
import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
@@ -62,6 +65,7 @@ import org.apache.flink.runtime.rest.handler.async.TriggerResponse;
import org.apache.flink.runtime.rest.messages.DashboardConfiguration;
import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobsOverviewHeaders;
import org.apache.flink.runtime.rest.messages.TriggerId;
import org.apache.flink.runtime.rest.messages.job.savepoints.SavepointDisposalRequest;
import org.apache.flink.runtime.rest.messages.job.savepoints.SavepointDisposalTriggerHeaders;
@@ -107,6 +111,7 @@ import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
@@ -114,6 +119,7 @@ import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
import static org.apache.flink.kubernetes.operator.config.FlinkConfigBuilder.FLINK_VERSION;
@@ -136,6 +142,32 @@ public class FlinkService {
4, new ExecutorThreadFactory("Flink-RestClusterClient-IO"));
}
+ private static List<JobStatusMessage> toJobStatusMessage(
+ MultipleJobsDetails multipleJobsDetails) {
+ return multipleJobsDetails.getJobs().stream()
+ .map(
+ details ->
+ new JobStatusMessage(
+ details.getJobId(),
+ details.getJobName(),
+ getEffectiveStatus(details),
+ details.getStartTime()))
+ .collect(Collectors.toList());
+ }
+
+ @VisibleForTesting
+ protected static JobStatus getEffectiveStatus(JobDetails details) {
+ int numRunning = details.getTasksPerState()[ExecutionState.RUNNING.ordinal()];
+ int numFinished = details.getTasksPerState()[ExecutionState.FINISHED.ordinal()];
+ boolean allRunningOrFinished = details.getNumTasks() == (numRunning + numFinished);
+ JobStatus effectiveStatus = details.getStatus();
+ if (JobStatus.RUNNING.equals(effectiveStatus) && !allRunningOrFinished) {
+ effectiveStatus = JobStatus.CREATED;
+ LOG.debug("Adjusting job state from {} to {}", JobStatus.RUNNING, effectiveStatus);
+ }
+ return effectiveStatus;
+ }
+
public KubernetesClient getKubernetesClient() {
return kubernetesClient;
}
@@ -329,14 +361,19 @@ public class FlinkService {
}
public Collection<JobStatusMessage> listJobs(Configuration conf) throws Exception {
- try (ClusterClient<String> clusterClient = getClusterClient(conf)) {
+ try (RestClusterClient<String> clusterClient =
+ (RestClusterClient<String>) getClusterClient(conf)) {
return clusterClient
- .listJobs()
+ .sendRequest(
+ JobsOverviewHeaders.getInstance(),
+ EmptyMessageParameters.getInstance(),
+ EmptyRequestBody.getInstance())
+ .thenApply(FlinkService::toJobStatusMessage)
.get(
configManager
.getOperatorConfiguration()
.getFlinkClientTimeout()
- .getSeconds(),
+ .toSeconds(),
TimeUnit.SECONDS);
}
}
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
index 0e29516..94ce385 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/TestingFlinkService.java
@@ -40,10 +40,15 @@ import org.apache.flink.kubernetes.operator.observer.SavepointFetchResult;
import org.apache.flink.kubernetes.operator.service.FlinkService;
import org.apache.flink.kubernetes.operator.utils.FlinkUtils;
import org.apache.flink.runtime.client.JobStatusMessage;
+import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.jobgraph.SavepointConfigOptions;
import org.apache.flink.runtime.jobmaster.JobResult;
import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.messages.webmonitor.JobDetails;
+import org.apache.flink.runtime.messages.webmonitor.MultipleJobsDetails;
import org.apache.flink.runtime.rest.messages.DashboardConfiguration;
+import org.apache.flink.runtime.rest.messages.EmptyResponseBody;
+import org.apache.flink.runtime.rest.messages.JobsOverviewHeaders;
import org.apache.flink.util.SerializedThrowable;
import io.fabric8.kubernetes.api.model.ObjectMeta;
@@ -296,9 +301,37 @@ public class TestingFlinkService extends FlinkService {
}
return CompletableFuture.completedFuture(builder.build());
});
+ clusterClient.setRequestProcessor(
+ (messageHeaders, messageParameters, requestBody) -> {
+ if (messageHeaders instanceof JobsOverviewHeaders) {
+ return CompletableFuture.completedFuture(getMultipleJobsDetails());
+ }
+ return CompletableFuture.completedFuture(EmptyResponseBody.getInstance());
+ });
return clusterClient;
}
+ private MultipleJobsDetails getMultipleJobsDetails() {
+ return new MultipleJobsDetails(
+ jobs.stream()
+ .map(tuple -> tuple.f1)
+ .map(TestingFlinkService::toJobDetails)
+ .collect(Collectors.toList()));
+ }
+
+ private static JobDetails toJobDetails(JobStatusMessage jobStatus) {
+ return new JobDetails(
+ jobStatus.getJobId(),
+ jobStatus.getJobName(),
+ jobStatus.getStartTime(),
+ -1,
+ System.currentTimeMillis() - jobStatus.getStartTime(),
+ jobStatus.getJobState(),
+ System.currentTimeMillis(),
+ new int[ExecutionState.values().length],
+ 0);
+ }
+
private String cancelJob(FlinkVersion flinkVersion, JobID jobID, boolean savepoint)
throws Exception {
Optional<Tuple2<String, JobStatusMessage>> jobOpt =
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/FlinkServiceTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/FlinkServiceTest.java
index 4b515aa..9390d7a 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/FlinkServiceTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/FlinkServiceTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.kubernetes.operator.service;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.client.program.ClusterClient;
import org.apache.flink.configuration.CheckpointingOptions;
@@ -35,7 +36,9 @@ import org.apache.flink.kubernetes.operator.crd.status.JobStatus;
import org.apache.flink.kubernetes.operator.crd.status.SavepointTriggerType;
import org.apache.flink.kubernetes.operator.exception.DeploymentFailedException;
import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
+import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.messages.Acknowledge;
+import org.apache.flink.runtime.messages.webmonitor.JobDetails;
import org.apache.flink.runtime.rest.handler.async.TriggerResponse;
import org.apache.flink.runtime.rest.messages.TriggerId;
import org.apache.flink.runtime.rest.messages.job.savepoints.SavepointTriggerMessageParameters;
@@ -53,6 +56,7 @@ import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.concurrent.CompletableFuture;
import static org.apache.flink.kubernetes.operator.config.FlinkConfigBuilder.FLINK_VERSION;
@@ -304,6 +308,64 @@ public class FlinkServiceTest {
objectMapper.readValue(flink15Response, CustomDashboardConfiguration.class);
}
+ @Test
+ public void testEffectiveStatus() {
+
+ JobDetails allRunning =
+ getJobDetails(
+ org.apache.flink.api.common.JobStatus.RUNNING,
+ Tuple2.of(ExecutionState.RUNNING, 4));
+ assertEquals(
+ org.apache.flink.api.common.JobStatus.RUNNING,
+ FlinkService.getEffectiveStatus(allRunning));
+
+ JobDetails allRunningOrFinished =
+ getJobDetails(
+ org.apache.flink.api.common.JobStatus.RUNNING,
+ Tuple2.of(ExecutionState.RUNNING, 2),
+ Tuple2.of(ExecutionState.FINISHED, 2));
+ assertEquals(
+ org.apache.flink.api.common.JobStatus.RUNNING,
+ FlinkService.getEffectiveStatus(allRunningOrFinished));
+
+ JobDetails allRunningOrScheduled =
+ getJobDetails(
+ org.apache.flink.api.common.JobStatus.RUNNING,
+ Tuple2.of(ExecutionState.RUNNING, 2),
+ Tuple2.of(ExecutionState.SCHEDULED, 2));
+ assertEquals(
+ org.apache.flink.api.common.JobStatus.CREATED,
+ FlinkService.getEffectiveStatus(allRunningOrScheduled));
+
+ JobDetails allFinished =
+ getJobDetails(
+ org.apache.flink.api.common.JobStatus.FINISHED,
+ Tuple2.of(ExecutionState.FINISHED, 4));
+ assertEquals(
+ org.apache.flink.api.common.JobStatus.FINISHED,
+ FlinkService.getEffectiveStatus(allFinished));
+ }
+
+ private JobDetails getJobDetails(
+ org.apache.flink.api.common.JobStatus status,
+ Tuple2<ExecutionState, Integer>... tasksPerState) {
+ int[] countPerState = new int[ExecutionState.values().length];
+ for (var taskPerState : tasksPerState) {
+ countPerState[taskPerState.f0.ordinal()] = taskPerState.f1;
+ }
+ int numTasks = Arrays.stream(countPerState).sum();
+ return new JobDetails(
+ new JobID(),
+ "test-job",
+ System.currentTimeMillis(),
+ -1,
+ 0,
+ status,
+ System.currentTimeMillis(),
+ countPerState,
+ numTasks);
+ }
+
private FlinkService createFlinkService(ClusterClient<String> clusterClient) {
return new FlinkService(client, new FlinkConfigManager(configuration)) {
@Override