improve class and variable naming

yunfengzhou-hub · yunfengzhou-hub · commit 5f93c07983c1 · 2025-09-03T17:59:43.000+08:00
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobDetailsHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobDetailsHandler.java
@@ -102,7 +102,8 @@ public Collection<ArchivedJson> archiveJsonWithPath(AccessExecutionGraph graph)
     }
 
     private static JobDetailsInfo createJobDetailsInfo(
-            AccessExecutionGraph executionGraph, @Nullable MetricStore.MetricStoreJobs jobs) {
+            AccessExecutionGraph executionGraph,
+            @Nullable MetricStore.JobMetricStoreSnapshot jobMetrics) {
         final long now = System.currentTimeMillis();
         final long startTime = executionGraph.getStatusTimestamp(JobStatus.INITIALIZING);
         final long endTime =
@@ -126,7 +127,7 @@ private static JobDetailsInfo createJobDetailsInfo(
                 executionGraph.getVerticesTopologically()) {
             final JobDetailsInfo.JobVertexDetailsInfo vertexDetailsInfo =
                     createJobVertexDetailsInfo(
-                            accessExecutionJobVertex, now, executionGraph.getJobID(), jobs);
+                            accessExecutionJobVertex, now, executionGraph.getJobID(), jobMetrics);
 
             jobVertexInfos.add(vertexDetailsInfo);
             jobVerticesPerState[vertexDetailsInfo.getExecutionState().ordinal()]++;
@@ -165,7 +166,10 @@ private static JobDetailsInfo createJobDetailsInfo(
     }
 
     private static JobDetailsInfo.JobVertexDetailsInfo createJobVertexDetailsInfo(
-            AccessExecutionJobVertex ejv, long now, JobID jobId, MetricStore.MetricStoreJobs jobs) {
+            AccessExecutionJobVertex ejv,
+            long now,
+            JobID jobId,
+            MetricStore.JobMetricStoreSnapshot jobMetrics) {
         int[] tasksPerState = new int[ExecutionState.values().length];
         long startTime = Long.MAX_VALUE;
         long endTime = 0;
@@ -216,7 +220,7 @@ private static JobDetailsInfo.JobVertexDetailsInfo createJobVertexDetailsInfo(
             // rather than the aggregation of all attempts.
             counts.addIOMetrics(
                     vertex.getCurrentExecutionAttempt(),
-                    jobs,
+                    jobMetrics,
                     jobId.toString(),
                     ejv.getJobVertexId().toString());
         }
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobVertexDetailsHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobVertexDetailsHandler.java
@@ -117,7 +117,7 @@ public Collection<ArchivedJson> archiveJsonWithPath(AccessExecutionGraph graph)
     private static JobVertexDetailsInfo createJobVertexDetailsInfo(
             AccessExecutionJobVertex jobVertex,
             JobID jobID,
-            @Nullable MetricStore.MetricStoreJobs jobs) {
+            @Nullable MetricStore.JobMetricStoreSnapshot jobMetrics) {
         List<SubtaskExecutionAttemptDetailsInfo> subtasks = new ArrayList<>();
         final long now = System.currentTimeMillis();
         for (AccessExecutionVertex vertex : jobVertex.getTaskVertices()) {
@@ -133,14 +133,14 @@ private static JobVertexDetailsInfo createJobVertexDetailsInfo(
                     if (attempt.getAttemptNumber() != execution.getAttemptNumber()) {
                         otherConcurrentAttempts.add(
                                 SubtaskExecutionAttemptDetailsInfo.create(
-                                        attempt, jobs, jobID, jobVertexID, null));
+                                        attempt, jobMetrics, jobID, jobVertexID, null));
                     }
                 }
             }
 
             subtasks.add(
                     SubtaskExecutionAttemptDetailsInfo.create(
-                            execution, jobs, jobID, jobVertexID, otherConcurrentAttempts));
+                            execution, jobMetrics, jobID, jobVertexID, otherConcurrentAttempts));
         }
 
         return new JobVertexDetailsInfo(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobVertexTaskManagersHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/JobVertexTaskManagersHandler.java
@@ -133,7 +133,7 @@ public Collection<ArchivedJson> archiveJsonWithPath(AccessExecutionGraph graph)
     private static JobVertexTaskManagersInfo createJobVertexTaskManagersInfo(
             AccessExecutionJobVertex jobVertex,
             JobID jobID,
-            @Nullable MetricStore.MetricStoreJobs jobs) {
+            @Nullable MetricStore.JobMetricStoreSnapshot jobMetrics) {
         // Build a map that groups task executions by TaskManager
         Map<TaskManagerLocation, List<AccessExecution>> taskManagerExecutions = new HashMap<>();
         Set<AccessExecution> representativeExecutions = new HashSet<>();
@@ -203,10 +203,16 @@ private static JobVertexTaskManagersInfo createJobVertexTaskManagersInfo(
                 endTime = Math.max(endTime, execution.getStateTimestamp(state));
 
                 counts.addIOMetrics(
-                        execution, jobs, jobID.toString(), jobVertex.getJobVertexId().toString());
+                        execution,
+                        jobMetrics,
+                        jobID.toString(),
+                        jobVertex.getJobVertexId().toString());
                 MutableIOMetrics current = new MutableIOMetrics();
                 current.addIOMetrics(
-                        execution, jobs, jobID.toString(), jobVertex.getJobVertexId().toString());
+                        execution,
+                        jobMetrics,
+                        jobID.toString(),
+                        jobVertex.getJobVertexId().toString());
                 ioMetricsInfos.add(
                         new IOMetricsInfo(
                                 current.getNumBytesIn(),
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/SubtaskCurrentAttemptDetailsHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/SubtaskCurrentAttemptDetailsHandler.java
@@ -89,19 +89,19 @@ protected SubtaskExecutionAttemptDetailsInfo handleRequest(
         List<SubtaskExecutionAttemptDetailsInfo> otherConcurrentAttempts = null;
 
         metricFetcher.update();
-        MetricStore.MetricStoreJobs jobs = metricFetcher.getMetricStore().getJobs();
+        MetricStore.JobMetricStoreSnapshot jobMetrics = metricFetcher.getMetricStore().getJobs();
         if (attempts.size() > 1) {
             otherConcurrentAttempts = new ArrayList<>();
             for (AccessExecution attempt : attempts) {
                 if (attempt.getAttemptNumber() != execution.getAttemptNumber()) {
                     otherConcurrentAttempts.add(
                             SubtaskExecutionAttemptDetailsInfo.create(
-                                    attempt, jobs, jobID, jobVertexID, null));
+                                    attempt, jobMetrics, jobID, jobVertexID, null));
                 }
             }
         }
 
         return SubtaskExecutionAttemptDetailsInfo.create(
-                execution, jobs, jobID, jobVertexID, otherConcurrentAttempts);
+                execution, jobMetrics, jobID, jobVertexID, otherConcurrentAttempts);
     }
 }
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/AggregatingSubtasksMetricsHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/AggregatingSubtasksMetricsHandler.java
@@ -76,7 +76,7 @@ Collection<? extends MetricStore.ComponentMetricStore> getStores(
         JobID jobID = request.getPathParameter(JobIDPathParameter.class);
         JobVertexID taskID = request.getPathParameter(JobVertexIdPathParameter.class);
 
-        MetricStore.MetricStoreJobs jobs = store.getJobs();
+        MetricStore.JobMetricStoreSnapshot jobMetrics = store.getJobs();
         Collection<String> subtaskRanges =
                 request.getQueryParameter(SubtasksFilterQueryParameter.class);
         if (subtaskRanges.isEmpty()) {
@@ -92,7 +92,8 @@ Collection<? extends MetricStore.ComponentMetricStore> getStores(
             Collection<MetricStore.ComponentMetricStore> subtaskStores = new ArrayList<>(8);
             for (int subtask : subtasks) {
                 MetricStore.ComponentMetricStore subtaskMetricStore =
-                        jobs.getSubtaskMetricStore(jobID.toString(), taskID.toString(), subtask);
+                        jobMetrics.getSubtaskMetricStore(
+                                jobID.toString(), taskID.toString(), subtask);
                 if (subtaskMetricStore != null) {
                     subtaskStores.add(subtaskMetricStore);
                 }
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/SubtaskMetricsHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/SubtaskMetricsHandler.java
@@ -40,7 +40,7 @@
 /**
  * Handler that returns subtask metrics.
  *
- * @see MetricStore.MetricStoreJobs#getSubtaskMetricStore (String, String, int)
+ * @see MetricStore.JobMetricStoreSnapshot#getSubtaskMetricStore (String, String, int)
  */
 public class SubtaskMetricsHandler extends AbstractMetricsHandler<SubtaskMetricsMessageParameters> {
 
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/metrics/MetricStore.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/metrics/MetricStore.java
@@ -239,8 +239,8 @@ public synchronized TaskMetricStore getTaskMetricStore(String jobID, String task
         return TaskMetricStore.unmodifiable(job.getTaskMetricStore(taskID));
     }
 
-    public synchronized MetricStoreJobs getJobs() {
-        return new MetricStoreJobs(unmodifiableMap(jobs));
+    public synchronized JobMetricStoreSnapshot getJobs() {
+        return new JobMetricStoreSnapshot(unmodifiableMap(jobs));
     }
 
     public synchronized Map<String, TaskManagerMetricStore> getTaskManagers() {
@@ -653,10 +653,10 @@ private static SubtaskMetricStore unmodifiable(SubtaskMetricStore source) {
 
     /** Sub-structure containing a snapshot of all jobs in the metric store. */
     @ThreadSafe
-    public static class MetricStoreJobs {
+    public static class JobMetricStoreSnapshot {
         private final Map<String, JobMetricStore> jobs;
 
-        private MetricStoreJobs(Map<String, JobMetricStore> jobs) {
+        private JobMetricStoreSnapshot(Map<String, JobMetricStore> jobs) {
             this.jobs = checkNotNull(jobs);
         }
 
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/util/MutableIOMetrics.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/util/MutableIOMetrics.java
@@ -72,13 +72,13 @@ public boolean isNumRecordsOutComplete() {
      * {@link MetricFetcher} is used to retrieve the required metrics.
      *
      * @param attempt Attempt whose IO metrics should be added
-     * @param jobs metrics for running jobs
+     * @param jobMetrics metrics for running jobs
      * @param jobID JobID to which the attempt belongs
      * @param taskID TaskID to which the attempt belongs
      */
     public void addIOMetrics(
             AccessExecution attempt,
-            @Nullable MetricStore.MetricStoreJobs jobs,
+            @Nullable MetricStore.JobMetricStoreSnapshot jobMetrics,
             String jobID,
             String taskID) {
         if (attempt.getState().isTerminal()) {
@@ -98,9 +98,9 @@ public void addIOMetrics(
                 }
             }
         } else { // execAttempt is still running, use MetricQueryService instead
-            if (jobs != null) {
+            if (jobMetrics != null) {
                 MetricStore.ComponentMetricStore metrics =
-                        jobs.getSubtaskAttemptMetricStore(
+                        jobMetrics.getSubtaskAttemptMetricStore(
                                 jobID,
                                 taskID,
                                 attempt.getParallelSubtaskIndex(),
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/SubtaskExecutionAttemptDetailsInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/SubtaskExecutionAttemptDetailsInfo.java
@@ -195,7 +195,7 @@ public List<SubtaskExecutionAttemptDetailsInfo> getOtherConcurrentAttempts() {
 
     public static SubtaskExecutionAttemptDetailsInfo create(
             AccessExecution execution,
-            @Nullable MetricStore.MetricStoreJobs jobs,
+            @Nullable MetricStore.JobMetricStoreSnapshot jobMetrics,
             JobID jobID,
             JobVertexID jobVertexID,
             @Nullable List<SubtaskExecutionAttemptDetailsInfo> otherConcurrentAttempts) {
@@ -215,7 +215,7 @@ public static SubtaskExecutionAttemptDetailsInfo create(
         final long duration = startTime > 0 ? ((endTime > 0 ? endTime : now) - startTime) : -1;
 
         final MutableIOMetrics ioMetrics = new MutableIOMetrics();
-        ioMetrics.addIOMetrics(execution, jobs, jobID.toString(), jobVertexID.toString());
+        ioMetrics.addIOMetrics(execution, jobMetrics, jobID.toString(), jobVertexID.toString());
 
         final IOMetricsInfo ioMetricsInfo =
                 new IOMetricsInfo(
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/metrics/MetricStoreTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/legacy/metrics/MetricStoreTest.java
@@ -50,7 +50,7 @@ class MetricStoreTest {
     @Test
     void testAdd() {
         MetricStore store = setupStore(new MetricStore());
-        MetricStore.MetricStoreJobs jobs = store.getJobs();
+        MetricStore.JobMetricStoreSnapshot jobMetrics = store.getJobs();
         assertThat(store.getJobManagerMetricStore().getMetric("abc.metric1", "-1")).isEqualTo("0");
         assertThat(store.getTaskManagerMetricStore("tmid").getMetric("abc.metric2", "-1"))
                 .isEqualTo("1");
@@ -64,15 +64,18 @@ void testAdd() {
                                 .getMetric("8.abc.metric5", "-1"))
                 .isEqualTo("14");
         assertThat(
-                        jobs.getSubtaskMetricStore(JOB_ID.toString(), "taskid", 8)
+                        jobMetrics
+                                .getSubtaskMetricStore(JOB_ID.toString(), "taskid", 8)
                                 .getMetric("abc.metric5", "-1"))
                 .isEqualTo("14");
         assertThat(
-                        jobs.getSubtaskAttemptMetricStore(JOB_ID.toString(), "taskid", 8, 1)
+                        jobMetrics
+                                .getSubtaskAttemptMetricStore(JOB_ID.toString(), "taskid", 8, 1)
                                 .getMetric("abc.metric5", "-1"))
                 .isEqualTo("4");
         assertThat(
-                        jobs.getSubtaskAttemptMetricStore(JOB_ID.toString(), "taskid", 8, 2)
+                        jobMetrics
+                                .getSubtaskAttemptMetricStore(JOB_ID.toString(), "taskid", 8, 2)
                                 .getMetric("abc.metric5", "-1"))
                 .isEqualTo("14");
 
@@ -89,20 +92,25 @@ void testAdd() {
                                 .getMetric("1.opname.abc.metric7", "-1"))
                 .isEqualTo("6");
         assertThat(
-                        jobs.getSubtaskMetricStore(JOB_ID.toString(), "taskid", 1)
+                        jobMetrics
+                                .getSubtaskMetricStore(JOB_ID.toString(), "taskid", 1)
                                 .getMetric("opname.abc.metric7", "-1"))
                 .isEqualTo("6");
-        assertThat(jobs.getSubtaskAttemptMetricStore(JOB_ID.toString(), "taskid", 1, 2)).isNull();
+        assertThat(jobMetrics.getSubtaskAttemptMetricStore(JOB_ID.toString(), "taskid", 1, 2))
+                .isNull();
         assertThat(
-                        jobs.getSubtaskAttemptMetricStore(JOB_ID.toString(), "taskid", 1, 3)
+                        jobMetrics
+                                .getSubtaskAttemptMetricStore(JOB_ID.toString(), "taskid", 1, 3)
                                 .getMetric("opname.abc.metric7", "-1"))
                 .isEqualTo("6");
         assertThat(
-                        jobs.getSubtaskAttemptMetricStore(JOB_ID.toString(), "taskid", 8, 2)
+                        jobMetrics
+                                .getSubtaskAttemptMetricStore(JOB_ID.toString(), "taskid", 8, 2)
                                 .getMetric("opname.abc.metric7", "-1"))
                 .isEqualTo("6");
         assertThat(
-                        jobs.getSubtaskAttemptMetricStore(JOB_ID.toString(), "taskid", 8, 4)
+                        jobMetrics
+                                .getSubtaskAttemptMetricStore(JOB_ID.toString(), "taskid", 8, 4)
                                 .getMetric("opname.abc.metric7", "-1"))
                 .isEqualTo("16");
 

Original file line number	Diff line number	Diff line change
`@@ -117,7 +117,7 @@ public Collection<ArchivedJson> archiveJsonWithPath(AccessExecutionGraph graph)`
`117`	`117`	`private static JobVertexDetailsInfo createJobVertexDetailsInfo(`
`118`	`118`	`AccessExecutionJobVertex jobVertex,`
`119`	`119`	`JobID jobID,`
`120`		`- @Nullable MetricStore.MetricStoreJobs jobs) {`
	`120`	`+ @Nullable MetricStore.JobMetricStoreSnapshot jobMetrics) {`
`121`	`121`	`List<SubtaskExecutionAttemptDetailsInfo> subtasks = new ArrayList<>();`
`122`	`122`	`final long now = System.currentTimeMillis();`
`123`	`123`	`for (AccessExecutionVertex vertex : jobVertex.getTaskVertices()) {`
`@@ -133,14 +133,14 @@ private static JobVertexDetailsInfo createJobVertexDetailsInfo(`
`133`	`133`	`if (attempt.getAttemptNumber() != execution.getAttemptNumber()) {`
`134`	`134`	`otherConcurrentAttempts.add(`
`135`	`135`	`SubtaskExecutionAttemptDetailsInfo.create(`
`136`		`- attempt, jobs, jobID, jobVertexID, null));`
	`136`	`+ attempt, jobMetrics, jobID, jobVertexID, null));`
`137`	`137`	`}`
`138`	`138`	`}`
`139`	`139`	`}`
`140`	`140`
`141`	`141`	`subtasks.add(`
`142`	`142`	`SubtaskExecutionAttemptDetailsInfo.create(`
`143`		`- execution, jobs, jobID, jobVertexID, otherConcurrentAttempts));`
	`143`	`+ execution, jobMetrics, jobID, jobVertexID, otherConcurrentAttempts));`
`144`	`144`	`}`
`145`	`145`
`146`	`146`	`return new JobVertexDetailsInfo(`
Original file line number	Diff line number	Diff line change
`@@ -89,19 +89,19 @@ protected SubtaskExecutionAttemptDetailsInfo handleRequest(`
`89`	`89`	`List<SubtaskExecutionAttemptDetailsInfo> otherConcurrentAttempts = null;`
`90`	`90`
`91`	`91`	`metricFetcher.update();`
`92`		`- MetricStore.MetricStoreJobs jobs = metricFetcher.getMetricStore().getJobs();`
	`92`	`+ MetricStore.JobMetricStoreSnapshot jobMetrics = metricFetcher.getMetricStore().getJobs();`
`93`	`93`	`if (attempts.size() > 1) {`
`94`	`94`	`otherConcurrentAttempts = new ArrayList<>();`
`95`	`95`	`for (AccessExecution attempt : attempts) {`
`96`	`96`	`if (attempt.getAttemptNumber() != execution.getAttemptNumber()) {`
`97`	`97`	`otherConcurrentAttempts.add(`
`98`	`98`	`SubtaskExecutionAttemptDetailsInfo.create(`
`99`		`- attempt, jobs, jobID, jobVertexID, null));`
	`99`	`+ attempt, jobMetrics, jobID, jobVertexID, null));`
`100`	`100`	`}`
`101`	`101`	`}`
`102`	`102`	`}`
`103`	`103`
`104`	`104`	`return SubtaskExecutionAttemptDetailsInfo.create(`
`105`		`- execution, jobs, jobID, jobVertexID, otherConcurrentAttempts);`
	`105`	`+ execution, jobMetrics, jobID, jobVertexID, otherConcurrentAttempts);`
`106`	`106`	`}`
`107`	`107`	`}`
Original file line number	Diff line number	Diff line change
`@@ -40,7 +40,7 @@`
`40`	`40`	`/**`
`41`	`41`	`* Handler that returns subtask metrics.`
`42`	`42`	`*`
`43`		`- * @see MetricStore.MetricStoreJobs#getSubtaskMetricStore (String, String, int)`
	`43`	`+ * @see MetricStore.JobMetricStoreSnapshot#getSubtaskMetricStore (String, String, int)`
`44`	`44`	`*/`
`45`	`45`	`public class SubtaskMetricsHandler extends AbstractMetricsHandler<SubtaskMetricsMessageParameters> {`
`46`	`46`
Original file line number	Diff line number	Diff line change
`@@ -239,8 +239,8 @@ public synchronized TaskMetricStore getTaskMetricStore(String jobID, String task`
`239`	`239`	`return TaskMetricStore.unmodifiable(job.getTaskMetricStore(taskID));`
`240`	`240`	`}`
`241`	`241`
`242`		`- public synchronized MetricStoreJobs getJobs() {`
`243`		`- return new MetricStoreJobs(unmodifiableMap(jobs));`
	`242`	`+ public synchronized JobMetricStoreSnapshot getJobs() {`
	`243`	`+ return new JobMetricStoreSnapshot(unmodifiableMap(jobs));`
`244`	`244`	`}`
`245`	`245`
`246`	`246`	`public synchronized Map<String, TaskManagerMetricStore> getTaskManagers() {`
`@@ -653,10 +653,10 @@ private static SubtaskMetricStore unmodifiable(SubtaskMetricStore source) {`
`653`	`653`
`654`	`654`	`/** Sub-structure containing a snapshot of all jobs in the metric store. */`
`655`	`655`	`@ThreadSafe`
`656`		`- public static class MetricStoreJobs {`
	`656`	`+ public static class JobMetricStoreSnapshot {`
`657`	`657`	`private final Map<String, JobMetricStore> jobs;`
`658`	`658`
`659`		`- private MetricStoreJobs(Map<String, JobMetricStore> jobs) {`
	`659`	`+ private JobMetricStoreSnapshot(Map<String, JobMetricStore> jobs) {`
`660`	`660`	`this.jobs = checkNotNull(jobs);`
`661`	`661`	`}`
`662`	`662`