This is an automated email from the ASF dual-hosted git repository.
XComp pushed a commit to branch release-1.20
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.20 by this push:
new 069ddfe5179 [FLINK-40003][runtime] Fix IOMetrics not visible to state
transition listeners
069ddfe5179 is described below
commit 069ddfe5179ec3e79632e21931ca80e775673f60
Author: Chris Johnson <[email protected]>
AuthorDate: Fri Jun 26 15:47:57 2026 -0500
[FLINK-40003][runtime] Fix IOMetrics not visible to state transition
listeners
Execution.markFinished() and processFail() updated accumulators and
IOMetrics after transitionState(), so ExecutionStateUpdateListeners
notified inline during the transition saw null from getIOMetrics().
Add transitionToTerminalState() with a preCompletionAction callback that
runs after the state is set but before terminalStateFuture is completed
and listeners are notified. The metrics/accumulators update is passed as
this callback so both terminalStateFuture consumers and listeners observe
it. Applied to markFinished(), processFail(), completeCancelling() and
recoverExecution().
Adapted for release-1.20:
DefaultExecutionGraph#registerExecutionStateUpdateListener
does not exist on this branch, so the test registers its listener at graph
construction time via TestingDefaultExecutionGraphBuilder instead of going
through a scheduler.
Co-Authored-By: Claude <[email protected]>
(cherry picked from commit c0f774fdcefab05cf25c986effaad78fa5ecff2a)
---
.../flink/runtime/executiongraph/Execution.java | 68 +++++++++---
.../DefaultExecutionGraphDeploymentTest.java | 116 +++++++++++++++++++++
2 files changed, 167 insertions(+), 17 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
index 46769a45f89..872732db36d 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
@@ -87,6 +87,7 @@ import static
org.apache.flink.runtime.execution.ExecutionState.FINISHED;
import static org.apache.flink.runtime.execution.ExecutionState.INITIALIZING;
import static org.apache.flink.runtime.execution.ExecutionState.RUNNING;
import static org.apache.flink.runtime.execution.ExecutionState.SCHEDULED;
+import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
import static org.apache.flink.util.Preconditions.checkState;
@@ -471,9 +472,12 @@ public class Execution
taskManagerLocationFuture.complete(location);
try {
- transitionState(this.state, FINISHED);
+ transitionToTerminalState(
+ this.state,
+ FINISHED,
+ null,
+ () -> updateAccumulatorsAndMetrics(userAccumulators,
metrics));
finishPartitionsAndUpdateConsumers();
- updateAccumulatorsAndMetrics(userAccumulators, metrics);
releaseAssignedResource(null);
vertex.getExecutionGraphAccessor().deregisterExecution(this);
} finally {
@@ -995,10 +999,13 @@ public class Execution
if (current == INITIALIZING || current == RUNNING || current ==
DEPLOYING) {
- if (transitionState(current, FINISHED)) {
+ if (transitionToTerminalState(
+ current,
+ FINISHED,
+ null,
+ () -> updateAccumulatorsAndMetrics(userAccumulators,
metrics))) {
try {
finishPartitionsAndUpdateConsumers();
- updateAccumulatorsAndMetrics(userAccumulators,
metrics);
releaseAssignedResource(null);
vertex.getExecutionGraphAccessor().deregisterExecution(this);
} finally {
@@ -1080,9 +1087,11 @@ public class Execution
|| current == INITIALIZING
|| current == DEPLOYING) {
- updateAccumulatorsAndMetrics(userAccumulators, metrics);
-
- if (transitionState(current, CANCELED)) {
+ if (transitionToTerminalState(
+ current,
+ CANCELED,
+ null,
+ () -> updateAccumulatorsAndMetrics(userAccumulators,
metrics))) {
finishCancellation(releasePartitions);
return;
}
@@ -1192,14 +1201,18 @@ public class Execution
return;
}
- checkState(transitionState(current, FAILED, t));
-
- // success (in a manner of speaking)
- this.failureCause =
- Optional.of(
- ErrorInfo.createErrorInfoWithNullableCause(t,
getStateTimestamp(FAILED)));
-
- updateAccumulatorsAndMetrics(userAccumulators, metrics);
+ checkState(
+ transitionToTerminalState(
+ current,
+ FAILED,
+ t,
+ () -> {
+ updateAccumulatorsAndMetrics(userAccumulators,
metrics);
+ this.failureCause =
+ Optional.of(
+
ErrorInfo.createErrorInfoWithNullableCause(
+ t,
getStateTimestamp(FAILED)));
+ }));
releaseAssignedResource(t);
vertex.getExecutionGraphAccessor().deregisterExecution(this);
@@ -1465,11 +1478,29 @@ public class Execution
}
private boolean transitionState(ExecutionState currentState,
ExecutionState targetState) {
- return transitionState(currentState, targetState, null);
+ return transitionState(currentState, targetState, null, null);
+ }
+
+ /**
+ * Transitions to a terminal state, running {@code preCompletionAction}
after the state is set
+ * but before {@link #terminalStateFuture} is completed and listeners are
notified. Use this for
+ * storing metrics/accumulators so that both {@code terminalStateFuture}
consumers and {@link
+ * ExecutionStateUpdateListener}s see them.
+ */
+ private boolean transitionToTerminalState(
+ ExecutionState currentState,
+ ExecutionState targetState,
+ @Nullable Throwable error,
+ Runnable preCompletionAction) {
+ checkArgument(targetState.isTerminal(), "targetState must be
terminal");
+ return transitionState(currentState, targetState, error,
preCompletionAction);
}
private boolean transitionState(
- ExecutionState currentState, ExecutionState targetState, Throwable
error) {
+ ExecutionState currentState,
+ ExecutionState targetState,
+ @Nullable Throwable error,
+ @Nullable Runnable preCompletionAction) {
// sanity check
if (currentState.isTerminal()) {
throw new IllegalStateException(
@@ -1505,6 +1536,9 @@ public class Execution
if (targetState == INITIALIZING || targetState == RUNNING) {
initializingOrRunningFuture.complete(null);
} else if (targetState.isTerminal()) {
+ if (preCompletionAction != null) {
+ preCompletionAction.run();
+ }
// complete the terminal state future
terminalStateFuture.complete(targetState);
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphDeploymentTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphDeploymentTest.java
index 5ba97ed9ae0..4614f30dfdb 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphDeploymentTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraphDeploymentTest.java
@@ -84,6 +84,7 @@ import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.atomic.AtomicReference;
import static org.assertj.core.api.Assertions.assertThat;
@@ -402,6 +403,121 @@ class DefaultExecutionGraphDeploymentTest {
assertThat(execution2.getUserAccumulators()).isEqualTo(accumulators);
}
+ @Test
+ void testMetricsVisibleToListenersOnMarkFinished() throws Exception {
+ testMetricsVisibleToListenersDuringTerminalTransition(
+ (execution, accumulators, metrics) -> {
+ // markFinished() only transitions from a running state.
+ execution.transitionState(ExecutionState.RUNNING);
+ execution.markFinished(accumulators, metrics);
+ });
+ }
+
+ @Test
+ void testMetricsVisibleToListenersOnMarkFailed() throws Exception {
+ testMetricsVisibleToListenersDuringTerminalTransition(
+ (execution, accumulators, metrics) ->
+ // fromSchedulerNg=true to reach the FAILED transition.
+ execution.markFailed(
+ new Exception("test failure"),
+ false,
+ accumulators,
+ metrics,
+ false,
+ true));
+ }
+
+ @Test
+ void testMetricsVisibleToListenersOnCompleteCancelling() throws Exception {
+ testMetricsVisibleToListenersDuringTerminalTransition(
+ (execution, accumulators, metrics) -> {
+ // completeCancelling() transitions from CANCELING to
CANCELED. The execution
+ // starts in CREATED, from which cancel() would jump
directly to CANCELED.
+ execution.transitionState(ExecutionState.RUNNING);
+ execution.cancel();
+ execution.completeCancelling(accumulators, metrics, true);
+ });
+ }
+
+ @Test
+ void testMetricsVisibleToListenersOnRecoverExecution() throws Exception {
+ testMetricsVisibleToListenersDuringTerminalTransition(
+ (execution, accumulators, metrics) ->
+ // recoverExecution() transitions directly to FINISHED
during JM failover
+ // recovery.
+ execution.recoverExecution(
+ execution.getAttemptId(),
+ new LocalTaskManagerLocation(),
+ accumulators,
+ metrics));
+ }
+
+ /**
+ * Verifies that IOMetrics and user accumulators are visible to {@link
+ * ExecutionStateUpdateListener}s at the time they are notified of a
terminal state transition.
+ */
+ private void testMetricsVisibleToListenersDuringTerminalTransition(
+ TerminalStateTransition terminalStateTransition) throws Exception {
+ JobVertex v1 = new JobVertex("v1", new JobVertexID());
+ v1.setParallelism(1);
+ v1.setInvokableClass(BatchTask.class);
+
+ // The listener receives an ExecutionAttemptID, not the Execution
object, and the terminal
+ // transition deregisters the execution. Capture the metrics and
accumulators directly from
+ // within the listener callback so we observe exactly what is visible
at notification time.
+ AtomicReference<IOMetrics> metricsSeenByListener = new
AtomicReference<>();
+ AtomicReference<Map<String, Accumulator<?, ?>>>
accumulatorsSeenByListener =
+ new AtomicReference<>();
+ AtomicReference<Execution> executionRef = new AtomicReference<>();
+
+ // On this branch the listener can only be wired at graph construction
time (master
+ // additionally has
DefaultExecutionGraph#registerExecutionStateUpdateListener), so the
+ // graph is built directly instead of through a scheduler.
+ DefaultExecutionGraph graph =
+ TestingDefaultExecutionGraphBuilder.newBuilder()
+ .setJobGraph(JobGraphTestUtils.streamingJobGraph(v1))
+ .setExecutionStateUpdateListener(
+ (id, previousState, newState) -> {
+ Execution execution = executionRef.get();
+ if (newState.isTerminal()
+ && execution != null
+ &&
id.equals(execution.getAttemptId())) {
+
metricsSeenByListener.set(execution.getIOMetrics());
+ accumulatorsSeenByListener.set(
+
execution.getUserAccumulators());
+ }
+ })
+ .build(EXECUTOR_EXTENSION.getExecutor());
+ graph.start(ComponentMainThreadExecutorServiceAdapter.forMainThread());
+
+ Execution execution =
graph.getRegisteredExecutions().values().iterator().next();
+ executionRef.set(execution);
+
+ IOMetrics ioMetrics = new IOMetrics(10, 20, 30, 40, 100, 200, 300);
+ Map<String, Accumulator<?, ?>> accumulators = new HashMap<>();
+ IntCounter counter = new IntCounter();
+ counter.add(42);
+ accumulators.put("acc", counter);
+
+ terminalStateTransition.transition(execution, accumulators, ioMetrics);
+
+ assertThat(metricsSeenByListener.get())
+ .as("IOMetrics visible to listener during terminal transition")
+ .isNotNull();
+ assertIOMetricsEqual(metricsSeenByListener.get(), ioMetrics);
+ assertThat(accumulatorsSeenByListener.get())
+ .as("User accumulators visible to listener during terminal
transition")
+ .isEqualTo(accumulators);
+ }
+
+ @FunctionalInterface
+ private interface TerminalStateTransition {
+ void transition(
+ Execution execution,
+ Map<String, Accumulator<?, ?>> accumulators,
+ IOMetrics metrics);
+ }
+
@Test
void testRegistrationOfExecutionsCanceled() throws Exception {
