[FLINK-35552][runtime] ExecutionGraph shouldn't expose CheckpointStat…

…sTracker. It's an implementation detail.
XComp · Jun 18, 2024 · 419bfc6 · 419bfc6
1 parent eb91358
commit 419bfc6
Show file tree

Hide file tree

Showing 4 changed files with 10 additions and 27 deletions.
diff --git a/...-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java b/...-runtime/src/main/java/org/apache/flink/runtime/executiongraph/DefaultExecutionGraph.java
@@ -559,12 +559,6 @@ public CheckpointCoordinator getCheckpointCoordinator() {
         return checkpointCoordinator;
     }
 
-    @Nullable
-    @Override
-    public CheckpointStatsTracker getCheckpointStatsTracker() {
-        return checkpointStatsTracker;
-    }
-
     @Override
     public KvStateLocationRegistry getKvStateLocationRegistry() {
         return kvStateLocationRegistry;

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionGraph.java
@@ -96,9 +96,6 @@ void enableCheckpointing(
     @Nullable
     CheckpointCoordinator getCheckpointCoordinator();
 
-    @Nullable
-    CheckpointStatsTracker getCheckpointStatsTracker();
-
     KvStateLocationRegistry getKvStateLocationRegistry();
 
     void setJsonPlan(String jsonPlan);

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/ExecutionGraphHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/ExecutionGraphHandler.java
@@ -35,6 +35,7 @@
 import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
 import org.apache.flink.runtime.jobgraph.IntermediateDataSetID;
 import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobgraph.tasks.CheckpointCoordinatorConfiguration;
 import org.apache.flink.runtime.jobmanager.PartitionProducerDisposedException;
 import org.apache.flink.runtime.jobmaster.SerializedInputSplit;
 import org.apache.flink.runtime.messages.checkpoint.AcknowledgeCheckpoint;
@@ -81,23 +82,20 @@ public void reportCheckpointMetrics(
     public void reportInitializationMetrics(
             ExecutionAttemptID executionAttemptId,
             SubTaskInitializationMetrics initializationMetrics) {
-        if (executionGraph.getCheckpointStatsTracker() == null) {
+        final CheckpointCoordinatorConfiguration checkpointConfig =
+                executionGraph.getCheckpointCoordinatorConfiguration();
+        if (checkpointConfig == null || !checkpointConfig.isCheckpointingEnabled()) {
             // TODO: Consider to support reporting initialization stats without checkpointing
             log.debug(
                     "Ignoring reportInitializationMetrics if checkpoint coordinator is not present");
             return;
         }
-        ioExecutor.execute(
-                () -> {
-                    try {
-                        executionGraph
-                                .getCheckpointStatsTracker()
-                                .reportInitializationMetrics(
-                                        executionAttemptId, initializationMetrics);
-                    } catch (Exception t) {
-                        log.warn("Error while reportInitializationMetrics", t);
-                    }
-                });
+
+        processCheckpointCoordinatorMessage(
+                "ReportInitializationMetrics",
+                coordinator ->
+                        coordinator.reportInitializationMetrics(
+                                executionAttemptId, initializationMetrics));
     }
 
     public void acknowledgeCheckpoint(

diff --git a/...est/java/org/apache/flink/runtime/scheduler/adaptive/StateTrackingMockExecutionGraph.java b/...est/java/org/apache/flink/runtime/scheduler/adaptive/StateTrackingMockExecutionGraph.java
@@ -294,12 +294,6 @@ public CheckpointCoordinator getCheckpointCoordinator() {
         throw new UnsupportedOperationException();
     }
 
-    @Nullable
-    @Override
-    public CheckpointStatsTracker getCheckpointStatsTracker() {
-        throw new UnsupportedOperationException();
-    }
-
     @Override
     public KvStateLocationRegistry getKvStateLocationRegistry() {
         throw new UnsupportedOperationException();