This is an automated email from the ASF dual-hosted git repository.
XComp pushed a commit to branch release-2.3
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-2.3 by this push:
new bda6d208c57 [FLINK-40003][runtime] Fix IOMetrics not visible to state
transition listeners
bda6d208c57 is described below
commit bda6d208c571641f75391f13b3089859174b426c
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().
Co-Authored-By: Claude <[email protected]>
(cherry picked from commit c0f774fdcefab05cf25c986effaad78fa5ecff2a)
---
.../flink/runtime/executiongraph/Execution.java | 68 +++++++++---
.../DefaultExecutionGraphDeploymentTest.java | 123 ++++++++++++++++++++-
2 files changed, 168 insertions(+), 23 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 a1301f604dc..a704e3290fd 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
@@ -95,6 +95,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;
@@ -486,9 +487,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 {
@@ -1166,10 +1170,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 {
@@ -1251,9 +1258,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;
}
@@ -1363,14 +1372,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);
@@ -1636,11 +1649,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(
@@ -1676,6 +1707,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 270f7f71c88..256a22cec5d 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.apache.flink.runtime.util.JobVertexConnectionUtils.connectNewDataSetAsInput;
import static org.assertj.core.api.Assertions.assertThat;
@@ -403,6 +404,106 @@ 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.
+ 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());
+
+ SchedulerBase scheduler = setupScheduler(v1, 1);
+ DefaultExecutionGraph graph = (DefaultExecutionGraph)
scheduler.getExecutionGraph();
+
+ Execution execution =
graph.getRegisteredExecutions().values().iterator().next();
+ ExecutionAttemptID attemptId = execution.getAttemptId();
+
+ // 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<>();
+ graph.registerExecutionStateUpdateListener(
+ (id, previousState, newState) -> {
+ if (newState.isTerminal() && id.equals(attemptId)) {
+ metricsSeenByListener.set(execution.getIOMetrics());
+
accumulatorsSeenByListener.set(execution.getUserAccumulators());
+ }
+ });
+
+ 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 {
@@ -501,20 +602,30 @@ class DefaultExecutionGraphDeploymentTest {
.isEqualTo(CheckpointingOptions.MAX_RETAINED_CHECKPOINTS.defaultValue().intValue());
}
+ private SchedulerBase setupScheduler(JobVertex v1, int dop1) throws
Exception {
+ return setupScheduler(new JobVertex[] {v1}, new int[] {dop1});
+ }
+
private SchedulerBase setupScheduler(JobVertex v1, int dop1, JobVertex v2,
int dop2)
throws Exception {
- v1.setParallelism(dop1);
- v2.setParallelism(dop2);
+ return setupScheduler(new JobVertex[] {v1, v2}, new int[] {dop1,
dop2});
+ }
- v1.setInvokableClass(BatchTask.class);
- v2.setInvokableClass(BatchTask.class);
+ private SchedulerBase setupScheduler(JobVertex[] vertices, int[]
parallelisms)
+ throws Exception {
+ int totalDop = 0;
+ for (int i = 0; i < vertices.length; i++) {
+ vertices[i].setParallelism(parallelisms[i]);
+ vertices[i].setInvokableClass(BatchTask.class);
+ totalDop += parallelisms[i];
+ }
DirectScheduledExecutorService executorService = new
DirectScheduledExecutorService();
// execution graph that executes actions synchronously
final SchedulerBase scheduler =
new DefaultSchedulerBuilder(
- JobGraphTestUtils.streamingJobGraph(v1, v2),
+ JobGraphTestUtils.streamingJobGraph(vertices),
ComponentMainThreadExecutorServiceAdapter.forMainThread(),
EXECUTOR_EXTENSION.getExecutor())
.setExecutionSlotAllocatorFactory(
@@ -530,7 +641,7 @@ class DefaultExecutionGraphDeploymentTest {
scheduler.startScheduling();
Map<ExecutionAttemptID, Execution> executions =
eg.getRegisteredExecutions();
- assertThat(executions).hasSize(dop1 + dop2);
+ assertThat(executions).hasSize(totalDop);
return scheduler;
}
