diff --git a/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/flinkcluster/FlinkClusterJobListFetcher.java b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/flinkcluster/FlinkClusterJobListFetcher.java
index e99424c..ba12cde 100644
--- a/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/flinkcluster/FlinkClusterJobListFetcher.java
+++ b/flink-autoscaler-standalone/src/main/java/org/apache/flink/autoscaler/standalone/flinkcluster/FlinkClusterJobListFetcher.java
@@ -20,12 +20,15 @@
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.autoscaler.JobAutoScalerContext;
 import org.apache.flink.autoscaler.standalone.JobListFetcher;
+import org.apache.flink.autoscaler.utils.JobStatusUtils;
 import org.apache.flink.client.program.rest.RestClusterClient;
 import org.apache.flink.configuration.Configuration;
 import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
 import org.apache.flink.runtime.client.JobStatusMessage;
+import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
 import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
 import org.apache.flink.runtime.rest.messages.JobMessageParameters;
+import org.apache.flink.runtime.rest.messages.JobsOverviewHeaders;
 import org.apache.flink.runtime.rest.messages.job.JobManagerJobConfigurationHeaders;
 import org.apache.flink.util.function.FunctionWithException;
 
@@ -54,7 +57,11 @@
     public Collection<JobAutoScalerContext<JobID>> fetch() throws Exception {
         try (var restClusterClient = restClientGetter.apply(new Configuration())) {
             return restClusterClient
-                    .listJobs()
+                    .sendRequest(
+                            JobsOverviewHeaders.getInstance(),
+                            EmptyMessageParameters.getInstance(),
+                            EmptyRequestBody.getInstance())
+                    .thenApply(JobStatusUtils::toJobStatusMessage)
                     .get(restClientTimeout.toSeconds(), TimeUnit.SECONDS)
                     .stream()
                     .map(
diff --git a/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/flinkcluster/FlinkClusterJobListFetcherTest.java b/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/flinkcluster/FlinkClusterJobListFetcherTest.java
index 658eb58..9192e52 100644
--- a/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/flinkcluster/FlinkClusterJobListFetcherTest.java
+++ b/flink-autoscaler-standalone/src/test/java/org/apache/flink/autoscaler/standalone/flinkcluster/FlinkClusterJobListFetcherTest.java
@@ -21,10 +21,13 @@
 import org.apache.flink.api.common.JobStatus;
 import org.apache.flink.client.program.rest.RestClusterClient;
 import org.apache.flink.configuration.Configuration;
-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.messages.webmonitor.JobDetails;
+import org.apache.flink.runtime.messages.webmonitor.MultipleJobsDetails;
 import org.apache.flink.runtime.rest.messages.ConfigurationInfo;
 import org.apache.flink.runtime.rest.messages.JobMessageParameters;
+import org.apache.flink.runtime.rest.messages.JobsOverviewHeaders;
 import org.apache.flink.runtime.rest.messages.MessageHeaders;
 import org.apache.flink.runtime.rest.messages.MessageParameters;
 import org.apache.flink.runtime.rest.messages.RequestBody;
@@ -38,13 +41,12 @@
 import javax.annotation.Nullable;
 
 import java.time.Duration;
-import java.time.Instant;
-import java.util.Collection;
-import java.util.List;
+import java.util.Arrays;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.stream.Collectors;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -56,12 +58,9 @@
     /** Test whether the job list and confs are expected. */
     @Test
     void testFetchJobListAndConfigurationInfo() throws Exception {
-        var job1 =
-                new JobStatusMessage(
-                        new JobID(), "", JobStatus.RUNNING, Instant.now().toEpochMilli());
-        var job2 =
-                new JobStatusMessage(
-                        new JobID(), "", JobStatus.CANCELLING, Instant.now().toEpochMilli());
+        var job1 = new JobID();
+        var job2 = new JobID();
+        var jobs = Map.of(job1, JobStatus.RUNNING, job2, JobStatus.CANCELLING);
 
         Configuration expectedConf1 = new Configuration();
         expectedConf1.setString("option_key1", "option_value1");
@@ -70,18 +69,17 @@
         expectedConf2.setString("option_key2", "option_value2");
         expectedConf2.setString("option_key3", "option_value3");
 
-        var jobs = Map.of(job1.getJobId(), job1, job2.getJobId(), job2);
-        var configurations = Map.of(job1.getJobId(), expectedConf1, job2.getJobId(), expectedConf2);
+        var configurations = Map.of(job1, expectedConf1, job2, expectedConf2);
         var closeCounter = new AtomicLong();
         FlinkClusterJobListFetcher jobListFetcher =
                 new FlinkClusterJobListFetcher(
                         getRestClusterClient(
-                                Either.Left(List.of(job1, job2)),
+                                Either.Left(jobs),
                                 Either.Left(
                                         Map.of(
-                                                job1.getJobId(),
+                                                job1,
                                                 ConfigurationInfo.from(expectedConf1),
-                                                job2.getJobId(),
+                                                job2,
                                                 ConfigurationInfo.from(expectedConf2))),
                                 closeCounter),
                         Duration.ofSeconds(10));
@@ -94,11 +92,9 @@
 
             assertThat(fetchedJobList).hasSize(2);
             for (var jobContext : fetchedJobList) {
-                JobStatusMessage expectedJobStatusMessage = jobs.get(jobContext.getJobID());
+                var expectedJobState = jobs.get(jobContext.getJobID());
                 Configuration expectedConf = configurations.get(jobContext.getJobID());
-                assertThat(expectedJobStatusMessage).isNotNull();
-                assertThat(jobContext.getJobStatus())
-                        .isEqualTo(expectedJobStatusMessage.getJobState());
+                assertThat(jobContext.getJobStatus()).isEqualTo(expectedJobState);
                 assertThat(jobContext.getConfiguration()).isNotNull().isEqualTo(expectedConf);
             }
         }
@@ -130,16 +126,13 @@
      */
     @Test
     void testFetchConfigurationException() {
-        var job1 =
-                new JobStatusMessage(
-                        new JobID(), "", JobStatus.RUNNING, Instant.now().toEpochMilli());
         var expectedException = new RuntimeException("Expected exception.");
         var closeCounter = new AtomicLong();
 
         FlinkClusterJobListFetcher jobListFetcher =
                 new FlinkClusterJobListFetcher(
                         getRestClusterClient(
-                                Either.Left(List.of(job1)),
+                                Either.Left(Map.of(new JobID(), JobStatus.RUNNING)),
                                 Either.Right(expectedException),
                                 closeCounter),
                         Duration.ofSeconds(10));
@@ -171,14 +164,12 @@
      */
     @Test
     void testFetchConfigurationTimeout() {
-        var job1 =
-                new JobStatusMessage(
-                        new JobID(), "", JobStatus.RUNNING, Instant.now().toEpochMilli());
         CompletableFuture<Void> closeFuture = new CompletableFuture<>();
 
         FlinkClusterJobListFetcher jobListFetcher =
                 new FlinkClusterJobListFetcher(
-                        getTimeoutableRestClusterClient(List.of(job1), null, closeFuture),
+                        getTimeoutableRestClusterClient(
+                                Map.of(new JobID(), JobStatus.RUNNING), null, closeFuture),
                         Duration.ofSeconds(2));
 
         assertThat(closeFuture).isNotDone();
@@ -189,8 +180,8 @@
     }
 
     /**
-     * @param jobListOrException When listJobs is called, return jobList if Either is left, return
-     *     failedFuture if Either is right.
+     * @param jobsOrException When the jobs overview is called, return jobList if Either is left,
+     *     return failedFuture if Either is right.
      * @param configurationsOrException When fetch job conf, return configuration if Either is left,
      *     return failedFuture if Either is right.
      * @param closeCounter Increment the count each time the {@link RestClusterClient#close} is
@@ -198,7 +189,7 @@
      */
     private static FunctionWithException<Configuration, RestClusterClient<String>, Exception>
             getRestClusterClient(
-                    Either<Collection<JobStatusMessage>, Throwable> jobListOrException,
+                    Either<Map<JobID, JobStatus>, Throwable> jobsOrException,
                     Either<Map<JobID, ConfigurationInfo>, Throwable> configurationsOrException,
                     AtomicLong closeCounter) {
         return conf ->
@@ -208,14 +199,6 @@
                         (c, e) -> new StandaloneClientHAServices("localhost")) {
 
                     @Override
-                    public CompletableFuture<Collection<JobStatusMessage>> listJobs() {
-                        if (jobListOrException.isLeft()) {
-                            return CompletableFuture.completedFuture(jobListOrException.left());
-                        }
-                        return CompletableFuture.failedFuture(jobListOrException.right());
-                    }
-
-                    @Override
                     public <
                                     M extends MessageHeaders<R, P, U>,
                                     U extends MessageParameters,
@@ -231,6 +214,22 @@
                             return (CompletableFuture<P>)
                                     CompletableFuture.completedFuture(
                                             configurationsOrException.left().get(jobID));
+                        } else if (h instanceof JobsOverviewHeaders) {
+                            if (jobsOrException.isLeft()) {
+                                return (CompletableFuture<P>)
+                                        CompletableFuture.completedFuture(
+                                                new MultipleJobsDetails(
+                                                        jobsOrException.left().entrySet().stream()
+                                                                .map(
+                                                                        entry ->
+                                                                                generateJobDetails(
+                                                                                        entry
+                                                                                                .getKey(),
+                                                                                        entry
+                                                                                                .getValue()))
+                                                                .collect(Collectors.toList())));
+                            }
+                            return CompletableFuture.failedFuture(jobsOrException.right());
                         }
                         fail("Unknown request");
                         return null;
@@ -245,7 +244,7 @@
     }
 
     /**
-     * @param jobList When listJobs is called, return jobList if it's not null, don't complete
+     * @param jobs When the jobs overview is called, return jobList if it's not null, don't complete
      *     future if it's null.
      * @param configuration When fetch job conf, return configuration if it's not null, don't
      *     complete future if it's null.
@@ -253,7 +252,7 @@
      */
     private static FunctionWithException<Configuration, RestClusterClient<String>, Exception>
             getTimeoutableRestClusterClient(
-                    @Nullable Collection<JobStatusMessage> jobList,
+                    @Nullable Map<JobID, JobStatus> jobs,
                     @Nullable ConfigurationInfo configuration,
                     CompletableFuture<Void> closeFuture) {
         return conf ->
@@ -263,14 +262,6 @@
                         (c, e) -> new StandaloneClientHAServices("localhost")) {
 
                     @Override
-                    public CompletableFuture<Collection<JobStatusMessage>> listJobs() {
-                        if (jobList == null) {
-                            return new CompletableFuture<>();
-                        }
-                        return CompletableFuture.completedFuture(jobList);
-                    }
-
-                    @Override
                     public <
                                     M extends MessageHeaders<R, P, U>,
                                     U extends MessageParameters,
@@ -283,6 +274,21 @@
                             }
                             return (CompletableFuture<P>)
                                     CompletableFuture.completedFuture(configuration);
+                        } else if (h instanceof JobsOverviewHeaders) {
+                            if (jobs == null) {
+                                return new CompletableFuture<>();
+                            }
+                            return (CompletableFuture<P>)
+                                    CompletableFuture.completedFuture(
+                                            new MultipleJobsDetails(
+                                                    jobs.entrySet().stream()
+                                                            .map(
+                                                                    entry ->
+                                                                            generateJobDetails(
+                                                                                    entry.getKey(),
+                                                                                    entry
+                                                                                            .getValue()))
+                                                            .collect(Collectors.toList())));
                         }
                         fail("Unknown request");
                         return null;
@@ -295,4 +301,25 @@
                     }
                 };
     }
+
+    private static JobDetails generateJobDetails(JobID jobID, JobStatus jobStatus) {
+        int[] countPerState = new int[ExecutionState.values().length];
+        if (jobStatus == JobStatus.RUNNING) {
+            countPerState[ExecutionState.RUNNING.ordinal()] = 5;
+            countPerState[ExecutionState.FINISHED.ordinal()] = 2;
+        } else if (jobStatus == JobStatus.CANCELLING) {
+            countPerState[ExecutionState.CANCELING.ordinal()] = 7;
+        }
+        int numTasks = Arrays.stream(countPerState).sum();
+        return new JobDetails(
+                jobID,
+                "test-job",
+                System.currentTimeMillis(),
+                -1,
+                0,
+                jobStatus,
+                System.currentTimeMillis(),
+                countPerState,
+                numTasks);
+    }
 }
diff --git a/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/utils/JobStatusUtils.java b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/utils/JobStatusUtils.java
new file mode 100644
index 0000000..c4b9923
--- /dev/null
+++ b/flink-autoscaler/src/main/java/org/apache/flink/autoscaler/utils/JobStatusUtils.java
@@ -0,0 +1,63 @@
+/*
+ * 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.flink.autoscaler.utils;
+
+import org.apache.flink.annotation.VisibleForTesting;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.runtime.client.JobStatusMessage;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.messages.webmonitor.JobDetails;
+import org.apache.flink.runtime.messages.webmonitor.MultipleJobsDetails;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** Job status related utilities. */
+public class JobStatusUtils {
+
+    private static final Logger LOG = LoggerFactory.getLogger(JobStatusUtils.class);
+
+    public 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
+    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) {
+            LOG.debug("Adjusting job state from {} to {}", JobStatus.RUNNING, JobStatus.CREATED);
+            return JobStatus.CREATED;
+        }
+        return effectiveStatus;
+    }
+}
diff --git a/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/utils/JobStatusUtilsTest.java b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/utils/JobStatusUtilsTest.java
new file mode 100644
index 0000000..4cdb166
--- /dev/null
+++ b/flink-autoscaler/src/test/java/org/apache/flink/autoscaler/utils/JobStatusUtilsTest.java
@@ -0,0 +1,77 @@
+/*
+ * 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.flink.autoscaler.utils;
+
+import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.messages.webmonitor.JobDetails;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Arrays;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/** Test for {@link JobStatusUtils}. */
+class JobStatusUtilsTest {
+
+    @Test
+    void effectiveStatusTest() {
+        var allRunning = getJobDetails(JobStatus.RUNNING, Tuple2.of(ExecutionState.RUNNING, 4));
+        assertEquals(JobStatus.RUNNING, JobStatusUtils.getEffectiveStatus(allRunning));
+
+        var allRunningOrFinished =
+                getJobDetails(
+                        JobStatus.RUNNING,
+                        Tuple2.of(ExecutionState.RUNNING, 2),
+                        Tuple2.of(ExecutionState.FINISHED, 2));
+        assertEquals(JobStatus.RUNNING, JobStatusUtils.getEffectiveStatus(allRunningOrFinished));
+
+        var allRunningOrScheduled =
+                getJobDetails(
+                        JobStatus.RUNNING,
+                        Tuple2.of(ExecutionState.RUNNING, 2),
+                        Tuple2.of(ExecutionState.SCHEDULED, 2));
+        assertEquals(JobStatus.CREATED, JobStatusUtils.getEffectiveStatus(allRunningOrScheduled));
+
+        var allFinished = getJobDetails(JobStatus.FINISHED, Tuple2.of(ExecutionState.FINISHED, 4));
+        assertEquals(JobStatus.FINISHED, JobStatusUtils.getEffectiveStatus(allFinished));
+    }
+
+    @SafeVarargs
+    private JobDetails getJobDetails(
+            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);
+    }
+}
diff --git a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
index 42ce5eb..ae8979a 100644
--- a/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
+++ b/flink-kubernetes-operator/src/main/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkService.java
@@ -21,6 +21,7 @@
 import org.apache.flink.api.common.JobID;
 import org.apache.flink.api.common.JobStatus;
 import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.autoscaler.utils.JobStatusUtils;
 import org.apache.flink.client.program.rest.RestClusterClient;
 import org.apache.flink.configuration.CheckpointingOptions;
 import org.apache.flink.configuration.Configuration;
@@ -50,12 +51,9 @@
 import org.apache.flink.kubernetes.operator.utils.EnvUtils;
 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;
@@ -272,7 +270,7 @@
                             JobsOverviewHeaders.getInstance(),
                             EmptyMessageParameters.getInstance(),
                             EmptyRequestBody.getInstance())
-                    .thenApply(AbstractFlinkService::toJobStatusMessage)
+                    .thenApply(JobStatusUtils::toJobStatusMessage)
                     .get(operatorConfig.getFlinkClientTimeout().toSeconds(), TimeUnit.SECONDS);
         }
     }
@@ -902,19 +900,6 @@
                 timeout);
     }
 
-    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 Configuration removeOperatorConfigs(Configuration config) {
         Configuration newConfig = new Configuration();
@@ -929,19 +914,6 @@
         return newConfig;
     }
 
-    @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;
-    }
-
     private void validateHaMetadataExists(Configuration conf) {
         if (!isHaMetadataAvailable(conf)) {
             throw new RecoveryFailureException(
diff --git a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
index 8a928a2..8e9683d 100644
--- a/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
+++ b/flink-kubernetes-operator/src/test/java/org/apache/flink/kubernetes/operator/service/AbstractFlinkServiceTest.java
@@ -52,10 +52,8 @@
 import org.apache.flink.kubernetes.operator.reconciler.ReconciliationUtils;
 import org.apache.flink.runtime.clusterframework.types.ResourceID;
 import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
-import org.apache.flink.runtime.execution.ExecutionState;
 import org.apache.flink.runtime.instance.HardwareDescription;
 import org.apache.flink.runtime.messages.Acknowledge;
-import org.apache.flink.runtime.messages.webmonitor.JobDetails;
 import org.apache.flink.runtime.rest.RestClient;
 import org.apache.flink.runtime.rest.handler.async.AsynchronousOperationResult;
 import org.apache.flink.runtime.rest.handler.async.TriggerResponse;
@@ -123,7 +121,6 @@
 import java.nio.file.Path;
 import java.time.Duration;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -962,63 +959,6 @@
     }
 
     @Test
-    public void effectiveStatusTest() {
-        JobDetails allRunning =
-                getJobDetails(
-                        org.apache.flink.api.common.JobStatus.RUNNING,
-                        Tuple2.of(ExecutionState.RUNNING, 4));
-        assertEquals(
-                org.apache.flink.api.common.JobStatus.RUNNING,
-                AbstractFlinkService.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,
-                AbstractFlinkService.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,
-                AbstractFlinkService.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,
-                AbstractFlinkService.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);
-    }
-
-    @Test
     public void isJobManagerReadyTest() throws Exception {
         AtomicReference<String> url = new AtomicReference<>();
         var clusterClient =