This is an automated email from the ASF dual-hosted git repository.
rmetzger pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 8ad4d1b5bc3 [FLINK-40602][runtime] Fail job listing and cluster
overview on JobMaster query error (#29136)
8ad4d1b5bc3 is described below
commit 8ad4d1b5bc325c2eee594b99d5dae01e949ead58
Author: Han You <[email protected]>
AuthorDate: Fri Oct 2 07:40:48 2026 -0500
[FLINK-40602][runtime] Fail job listing and cluster overview on JobMaster
query error (#29136)
The Dispatcher queries every registered JobMaster to build the job list
and the cluster overview, but silently drops any JobMaster whose query
fails or times out. The request still succeeds, so clients of
/jobs/overview (e.g. the Kubernetes operator) conclude the job is gone.
This commit fails the whole request instead, with an exception naming
the job whose JobMaster could not be queried.
Co-authored-by: Han You <[email protected]>
---
.../flink/runtime/dispatcher/Dispatcher.java | 52 ++++++---------
.../flink/runtime/dispatcher/DispatcherTest.java | 78 ++++++++++++++++++++++
.../runtime/jobmaster/TestingJobManagerRunner.java | 38 +++++++++--
3 files changed, 131 insertions(+), 37 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
index c29be3274f2..50322989af1 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/Dispatcher.java
@@ -1595,15 +1595,10 @@ public abstract class Dispatcher extends
FencedRpcEndpoint<DispatcherId>
resourceManagerGateway ->
resourceManagerGateway.requestResourceOverview(timeout));
- final List<CompletableFuture<Optional<JobStatus>>>
optionalJobInformation =
- queryJobMastersForInformation(
- jobManagerRunner ->
jobManagerRunner.requestJobStatus(timeout));
-
- CompletableFuture<Collection<Optional<JobStatus>>>
allOptionalJobsFuture =
- FutureUtils.combineAll(optionalJobInformation);
-
CompletableFuture<Collection<JobStatus>> allJobsFuture =
-
allOptionalJobsFuture.thenApply(this::flattenOptionalCollection);
+ FutureUtils.combineAll(
+ queryJobMastersForInformation(
+ jobManagerRunner ->
jobManagerRunner.requestJobStatus(timeout)));
final JobsOverview completedJobsOverview = getCompletedJobsOverview();
@@ -1628,15 +1623,10 @@ public abstract class Dispatcher extends
FencedRpcEndpoint<DispatcherId>
@Override
public CompletableFuture<MultipleJobsDetails>
requestMultipleJobDetails(Duration timeout) {
- List<CompletableFuture<Optional<JobDetails>>>
individualOptionalJobDetails =
- queryJobMastersForInformation(
- jobManagerRunner ->
jobManagerRunner.requestJobDetails(timeout));
-
- CompletableFuture<Collection<Optional<JobDetails>>>
optionalCombinedJobDetails =
- FutureUtils.combineAll(individualOptionalJobDetails);
-
CompletableFuture<Collection<JobDetails>> combinedJobDetails =
-
optionalCombinedJobDetails.thenApply(this::flattenOptionalCollection);
+ FutureUtils.combineAll(
+ queryJobMastersForInformation(
+ jobManagerRunner ->
jobManagerRunner.requestJobDetails(timeout)));
final Collection<JobDetails> completedJobDetails =
getCompletedJobDetails();
@@ -2491,28 +2481,30 @@ public abstract class Dispatcher extends
FencedRpcEndpoint<DispatcherId>
.thenCompose(Function.identity());
}
- private <T> List<T> flattenOptionalCollection(Collection<Optional<T>>
optionalCollection) {
- return optionalCollection.stream()
- .filter(Optional::isPresent)
- .map(Optional::get)
- .collect(Collectors.toList());
- }
-
@Nonnull
- private <T> List<CompletableFuture<Optional<T>>>
queryJobMastersForInformation(
+ private <T> List<CompletableFuture<T>> queryJobMastersForInformation(
Function<JobManagerRunner, CompletableFuture<T>> queryFunction) {
- List<CompletableFuture<Optional<T>>> optionalJobInformation =
+ List<CompletableFuture<T>> jobInformation =
new ArrayList<>(jobManagerRunnerRegistry.size());
for (JobManagerRunner job :
jobManagerRunnerRegistry.getJobManagerRunners()) {
- final CompletableFuture<Optional<T>> queryResult =
+ final CompletableFuture<T> queryResult =
queryFunction
.apply(job)
- .handle((T value, Throwable t) ->
Optional.ofNullable(value));
- optionalJobInformation.add(queryResult);
- }
- return optionalJobInformation;
+ .exceptionally(
+ throwable -> {
+ throw new CompletionException(
+ new FlinkException(
+ String.format(
+ "Could not
retrieve information about job %s from its JobMaster.",
+
job.getJobID()),
+
ExceptionUtils.stripCompletionException(
+ throwable)));
+ });
+ jobInformation.add(queryResult);
+ }
+ return jobInformation;
}
private CompletableFuture<Void> waitForTerminatingJob(
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
index 1c85c7c3b2c..680655e5551 100755
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/dispatcher/DispatcherTest.java
@@ -146,6 +146,7 @@ import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
@@ -1407,6 +1408,83 @@ public class DispatcherTest extends
AbstractDispatcherTest {
Stream.of(jobId,
secondJobID).sorted().collect(Collectors.toList()));
}
+ /**
+ * A JobMaster that fails or times out on {@code requestJobDetails} must
not cause its running
+ * job to be silently omitted from an otherwise successful response:
clients (such as the
+ * Kubernetes operator) treat absence from this list as "job not found".
+ */
+ @Test
+ public void
testRequestMultipleJobDetails_doesNotSilentlyOmitJobWhoseJobMasterQueryFails()
+ throws Exception {
+ final JobID secondJobID = new JobID();
+ final DispatcherGateway dispatcherGateway =
+ createDispatcherWithUnresponsiveSecondJob(
+ TestingJobManagerRunner.newBuilder()
+ .setJobId(secondJobID)
+ .setJobDetailsFutureFunction(
+
DispatcherTest::failedJobMasterQueryFuture));
+
+ assertFailsWithJobMasterQueryFailure(
+ dispatcherGateway.requestMultipleJobDetails(TIMEOUT),
secondJobID);
+ }
+
+ /**
+ * A JobMaster that fails or times out on {@code requestJobStatus} must
not cause its running
+ * job to be silently left out of the job counts of an otherwise
successful response.
+ */
+ @Test
+ public void
testRequestClusterOverview_doesNotSilentlyOmitJobWhoseJobMasterQueryFails()
+ throws Exception {
+ final JobID secondJobID = new JobID();
+ final DispatcherGateway dispatcherGateway =
+ createDispatcherWithUnresponsiveSecondJob(
+ TestingJobManagerRunner.newBuilder()
+ .setJobId(secondJobID)
+
.setJobStatusFunction(DispatcherTest::failedJobMasterQueryFuture));
+
+ assertFailsWithJobMasterQueryFailure(
+ dispatcherGateway.requestClusterOverview(TIMEOUT),
secondJobID);
+ }
+
+ /**
+ * Fails through {@code thenApply}, like the production {@code
JobManagerRunner} queries do, so
+ * that the failure arrives wrapped in a {@link CompletionException}.
+ */
+ private static <T> CompletableFuture<T> failedJobMasterQueryFuture() {
+ return FutureUtils.<T>completedExceptionally(
+ new TimeoutException("JobMaster did not answer in
time"))
+ .thenApply(Function.identity());
+ }
+
+ private DispatcherGateway createDispatcherWithUnresponsiveSecondJob(
+ TestingJobManagerRunner.Builder unresponsiveJobManagerRunner)
throws Exception {
+ final TestingJobManagerRunner secondJobManagerRunner =
unresponsiveJobManagerRunner.build();
+ final JobGraph secondJobGraph = JobGraphTestUtils.streamingJobGraph();
+ secondJobGraph.setJobID(secondJobManagerRunner.getJobID());
+ secondJobGraph.setApplicationId(applicationId);
+ final JobManagerRunnerFactory jobManagerRunnerFactory =
+ new QueuedJobManagerRunnerFactory(
+
runningJobManagerRunnerWithJobStatus(JobStatus.RUNNING, jobId, 10L),
+ secondJobManagerRunner);
+
+ return createDispatcherAndStartJobs(
+ jobManagerRunnerFactory, Arrays.asList(jobGraph,
secondJobGraph));
+ }
+
+ private static void assertFailsWithJobMasterQueryFailure(
+ CompletableFuture<?> future, JobID failedJobId) {
+ assertThatFuture(future)
+ .eventuallyFailsWith(ExecutionException.class)
+ .havingCause()
+ .isInstanceOf(FlinkException.class)
+ .withMessage(
+ "Could not retrieve information about job %s from its
JobMaster.",
+ failedJobId)
+ .havingCause()
+ .isInstanceOf(TimeoutException.class)
+ .withMessage("JobMaster did not answer in time");
+ }
+
@Test
public void testRequestMultipleJobDetails_isSerializable() throws
Exception {
final JobManagerRunnerFactory blockingJobMaster =
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingJobManagerRunner.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingJobManagerRunner.java
index abb2c1b5584..e58dca76d9c 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingJobManagerRunner.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/TestingJobManagerRunner.java
@@ -28,6 +28,8 @@ import
org.apache.flink.runtime.messages.webmonitor.JobDetails;
import org.apache.flink.runtime.scheduler.ExecutionGraphInfo;
import org.apache.flink.util.Preconditions;
+import javax.annotation.Nullable;
+
import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;
@@ -49,7 +51,9 @@ public class TestingJobManagerRunner implements
JobManagerRunner {
private final CompletableFuture<JobManagerRunnerResult> resultFuture;
- private final Supplier<JobDetails> jobDetailsFunction;
+ private final Supplier<CompletableFuture<JobDetails>> jobDetailsFunction;
+
+ private final Supplier<CompletableFuture<JobStatus>> jobStatusFunction;
private final OneShotLatch closeAsyncCalledLatch = new OneShotLatch();
@@ -60,12 +64,17 @@ public class TestingJobManagerRunner implements
JobManagerRunner {
boolean blockingTermination,
CompletableFuture<JobMasterGateway> jobMasterGatewayFuture,
CompletableFuture<JobManagerRunnerResult> resultFuture,
- Supplier<JobDetails> jobDetailsFunction) {
+ Supplier<CompletableFuture<JobDetails>> jobDetailsFunction,
+ @Nullable Supplier<CompletableFuture<JobStatus>>
jobStatusFunction) {
this.jobId = jobId;
this.blockingTermination = blockingTermination;
this.jobMasterGatewayFuture = jobMasterGatewayFuture;
this.resultFuture = resultFuture;
this.jobDetailsFunction = jobDetailsFunction;
+ this.jobStatusFunction =
+ jobStatusFunction != null
+ ? jobStatusFunction
+ : () ->
CompletableFuture.completedFuture(this.jobStatus);
this.terminationFuture = new CompletableFuture<>();
final ExecutionGraphInfo suspendedExecutionGraphInfo =
@@ -111,12 +120,12 @@ public class TestingJobManagerRunner implements
JobManagerRunner {
@Override
public CompletableFuture<JobStatus> requestJobStatus(Duration timeout) {
- return CompletableFuture.completedFuture(jobStatus);
+ return jobStatusFunction.get();
}
@Override
public CompletableFuture<JobDetails> requestJobDetails(Duration timeout) {
- return CompletableFuture.completedFuture(jobDetailsFunction.get());
+ return jobDetailsFunction.get();
}
@Override
@@ -183,10 +192,11 @@ public class TestingJobManagerRunner implements
JobManagerRunner {
private CompletableFuture<JobMasterGateway> jobMasterGatewayFuture =
new CompletableFuture<>();
private CompletableFuture<JobManagerRunnerResult> resultFuture = new
CompletableFuture<>();
- private Supplier<JobDetails> jobDetailsFunction =
+ private Supplier<CompletableFuture<JobDetails>> jobDetailsFunction =
() -> {
throw new UnsupportedOperationException();
};
+ @Nullable private Supplier<CompletableFuture<JobStatus>>
jobStatusFunction = null;
private Builder() {
// No-op.
@@ -216,7 +226,20 @@ public class TestingJobManagerRunner implements
JobManagerRunner {
}
public Builder setJobDetailsFunction(Supplier<JobDetails>
jobDetailsFunction) {
- this.jobDetailsFunction =
Preconditions.checkNotNull(jobDetailsFunction);
+ Preconditions.checkNotNull(jobDetailsFunction);
+ return setJobDetailsFutureFunction(
+ () ->
CompletableFuture.completedFuture(jobDetailsFunction.get()));
+ }
+
+ public Builder setJobDetailsFutureFunction(
+ Supplier<CompletableFuture<JobDetails>>
jobDetailsFutureFunction) {
+ this.jobDetailsFunction =
Preconditions.checkNotNull(jobDetailsFutureFunction);
+ return this;
+ }
+
+ public Builder setJobStatusFunction(
+ Supplier<CompletableFuture<JobStatus>> jobStatusFunction) {
+ this.jobStatusFunction =
Preconditions.checkNotNull(jobStatusFunction);
return this;
}
@@ -227,7 +250,8 @@ public class TestingJobManagerRunner implements
JobManagerRunner {
blockingTermination,
jobMasterGatewayFuture,
resultFuture,
- jobDetailsFunction);
+ jobDetailsFunction,
+ jobStatusFunction);
}
}
}
