You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2017/10/11 12:43:10 UTC
[2/2] flink git commit: [FLINK-7707] [flip6] Add
TaskCheckpointStatisticDetailsHandler for new REST endpoint
[FLINK-7707] [flip6] Add TaskCheckpointStatisticDetailsHandler for new REST endpoint
This closes #4772.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/bc4638a3
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/bc4638a3
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/bc4638a3
Branch: refs/heads/master
Commit: bc4638a3c96049de3ef615159cf83bbd88019575
Parents: 7b99670
Author: Till Rohrmann <tr...@apache.org>
Authored: Wed Oct 4 14:41:54 2017 +0200
Committer: Till Rohrmann <tr...@apache.org>
Committed: Wed Oct 11 14:41:20 2017 +0200
----------------------------------------------------------------------
.../webmonitor/history/HistoryServer.java | 2 +-
.../dispatcher/DispatcherRestEndpoint.java | 18 +-
.../job/AbstractExecutionGraphHandler.java | 9 +
.../checkpoints/AbstractCheckpointHandler.java | 18 +-
.../CheckpointStatisticDetailsHandler.java | 5 +-
.../CheckpointingStatisticsHandler.java | 7 +-
.../TaskCheckpointStatisticDetailsHandler.java | 151 ++++++++++
.../handler/legacy/ClusterConfigHandler.java | 4 +-
.../handler/legacy/ClusterOverviewHandler.java | 2 +-
.../handler/legacy/DashboardConfigHandler.java | 2 +-
.../checkpoints/CheckpointStatsHandler.java | 7 +-
.../messages/ClusterConfigurationInfo.java | 59 ----
.../messages/ClusterConfigurationInfoEntry.java | 81 ------
.../legacy/messages/DashboardConfiguration.java | 137 ---------
.../messages/StatusOverviewWithVersion.java | 130 ---------
.../rest/messages/ClusterConfigurationInfo.java | 58 ++++
.../messages/ClusterConfigurationInfoEntry.java | 81 ++++++
.../ClusterConfigurationInfoHeaders.java | 1 -
.../rest/messages/ClusterOverviewHeaders.java | 1 -
.../rest/messages/DashboardConfiguration.java | 136 +++++++++
.../messages/DashboardConfigurationHeaders.java | 1 -
.../rest/messages/JobVertexIdPathParameter.java | 43 +++
.../messages/StatusOverviewWithVersion.java | 129 +++++++++
.../checkpoints/CheckpointStatistics.java | 100 +------
.../checkpoints/CheckpointingStatistics.java | 62 ----
.../checkpoints/MinMaxAvgStatistics.java | 86 ++++++
.../SubtaskCheckpointStatistics.java | 283 +++++++++++++++++++
.../TaskCheckpointMessageParameters.java | 40 +++
.../checkpoints/TaskCheckpointStatistics.java | 151 ++++++++++
.../TaskCheckpointStatisticsHeaders.java | 72 +++++
...kCheckpointStatisticsWithSubtaskDetails.java | 278 ++++++++++++++++++
.../legacy/DashboardConfigHandlerTest.java | 2 +-
.../messages/CheckpointConfigInfoTest.java | 45 ---
.../messages/CheckpointingStatisticsTest.java | 134 ---------
.../messages/ClusterConfigurationInfoTest.java | 39 ---
.../messages/DashboardConfigurationTest.java | 40 ---
.../legacy/messages/JobConfigInfoTest.java | 52 ----
.../legacy/messages/JobPlanInfoTest.java | 1 +
.../RestRequestMarshallingTestBase.java | 63 -----
.../RestResponseMarshallingTestBase.java | 63 -----
.../messages/StatusOverviewWithVersionTest.java | 44 ---
.../messages/BlobServerPortResponseTest.java | 2 -
.../messages/ClusterConfigurationInfoTest.java | 39 +++
.../messages/DashboardConfigurationTest.java | 40 +++
.../rest/messages/JobConfigInfoTest.java | 51 ++++
.../rest/messages/JobSubmitRequestBodyTest.java | 1 -
.../messages/JobSubmitResponseBodyTest.java | 1 -
.../RestRequestMarshallingTestBase.java | 62 ++++
.../RestResponseMarshallingTestBase.java | 62 ++++
.../messages/StatusOverviewWithVersionTest.java | 44 +++
.../checkpoints/CheckpointConfigInfoTest.java | 45 +++
.../CheckpointingStatisticsTest.java | 137 +++++++++
.../TaskCheckpointStatisticsTest.java | 46 +++
...ckpointStatisticsWithSubtaskDetailsTest.java | 72 +++++
.../SavepointTriggerResponseBodyTest.java | 2 +-
55 files changed, 2164 insertions(+), 1077 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java
----------------------------------------------------------------------
diff --git a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java
index 4a28182..9c3b51e 100644
--- a/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java
+++ b/flink-runtime-web/src/main/java/org/apache/flink/runtime/webmonitor/history/HistoryServer.java
@@ -28,7 +28,7 @@ import org.apache.flink.core.fs.Path;
import org.apache.flink.runtime.history.FsJobArchivist;
import org.apache.flink.runtime.net.SSLUtils;
import org.apache.flink.runtime.rest.handler.legacy.DashboardConfigHandler;
-import org.apache.flink.runtime.rest.handler.legacy.messages.DashboardConfiguration;
+import org.apache.flink.runtime.rest.messages.DashboardConfiguration;
import org.apache.flink.runtime.security.SecurityConfiguration;
import org.apache.flink.runtime.security.SecurityUtils;
import org.apache.flink.runtime.webmonitor.WebMonitorUtils;
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java
index 6297b41..d2c1428 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/dispatcher/DispatcherRestEndpoint.java
@@ -36,6 +36,7 @@ import org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointConfigHan
import org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatisticDetailsHandler;
import org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointStatsCache;
import org.apache.flink.runtime.rest.handler.job.checkpoints.CheckpointingStatisticsHandler;
+import org.apache.flink.runtime.rest.handler.job.checkpoints.TaskCheckpointStatisticDetailsHandler;
import org.apache.flink.runtime.rest.handler.legacy.ClusterConfigHandler;
import org.apache.flink.runtime.rest.handler.legacy.ClusterOverviewHandler;
import org.apache.flink.runtime.rest.handler.legacy.CurrentJobsOverviewHandler;
@@ -43,20 +44,21 @@ import org.apache.flink.runtime.rest.handler.legacy.DashboardConfigHandler;
import org.apache.flink.runtime.rest.handler.legacy.ExecutionGraphCache;
import org.apache.flink.runtime.rest.handler.legacy.files.StaticFileServerHandler;
import org.apache.flink.runtime.rest.handler.legacy.files.WebContentHandlerSpecification;
-import org.apache.flink.runtime.rest.handler.legacy.messages.ClusterConfigurationInfo;
-import org.apache.flink.runtime.rest.handler.legacy.messages.DashboardConfiguration;
-import org.apache.flink.runtime.rest.handler.legacy.messages.StatusOverviewWithVersion;
+import org.apache.flink.runtime.rest.messages.ClusterConfigurationInfo;
import org.apache.flink.runtime.rest.messages.ClusterConfigurationInfoHeaders;
import org.apache.flink.runtime.rest.messages.ClusterOverviewHeaders;
import org.apache.flink.runtime.rest.messages.CurrentJobsOverviewHandlerHeaders;
+import org.apache.flink.runtime.rest.messages.DashboardConfiguration;
import org.apache.flink.runtime.rest.messages.DashboardConfigurationHeaders;
import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
import org.apache.flink.runtime.rest.messages.JobConfigHeaders;
import org.apache.flink.runtime.rest.messages.JobPlanHeaders;
import org.apache.flink.runtime.rest.messages.JobTerminationHeaders;
+import org.apache.flink.runtime.rest.messages.StatusOverviewWithVersion;
import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointConfigHeaders;
import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointStatisticDetailsHeaders;
import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointingStatisticsHeaders;
+import org.apache.flink.runtime.rest.messages.checkpoints.TaskCheckpointStatisticsHeaders;
import org.apache.flink.runtime.webmonitor.WebMonitorUtils;
import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
import org.apache.flink.util.FileUtils;
@@ -196,6 +198,15 @@ public class DispatcherRestEndpoint extends RestServerEndpoint {
executionGraphCache,
executor);
+ TaskCheckpointStatisticDetailsHandler taskCheckpointStatisticDetailsHandler = new TaskCheckpointStatisticDetailsHandler(
+ restAddressFuture,
+ leaderRetriever,
+ timeout,
+ TaskCheckpointStatisticsHeaders.getInstance(),
+ executionGraphCache,
+ executor,
+ checkpointStatsCache);
+
final File tmpDir = restConfiguration.getTmpDir();
Optional<StaticFileServerHandler<DispatcherGateway>> optWebContent;
@@ -221,6 +232,7 @@ public class DispatcherRestEndpoint extends RestServerEndpoint {
handlers.add(Tuple2.of(CheckpointingStatisticsHeaders.getInstance(), checkpointStatisticsHandler));
handlers.add(Tuple2.of(CheckpointStatisticDetailsHeaders.getInstance(), checkpointStatisticDetailsHandler));
handlers.add(Tuple2.of(JobPlanHeaders.getInstance(), jobPlanHandler));
+ handlers.add(Tuple2.of(TaskCheckpointStatisticsHeaders.getInstance(), taskCheckpointStatisticDetailsHandler));
BlobServerPortHandler blobServerPortHandler = new BlobServerPortHandler(restAddressFuture, leaderRetriever, timeout);
handlers.add(Tuple2.of(blobServerPortHandler.getMessageHeaders(), blobServerPortHandler));
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/AbstractExecutionGraphHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/AbstractExecutionGraphHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/AbstractExecutionGraphHandler.java
index 5348b55..63c3e35 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/AbstractExecutionGraphHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/AbstractExecutionGraphHandler.java
@@ -81,5 +81,14 @@ public abstract class AbstractExecutionGraphHandler<R extends ResponseBody, M ex
executor);
}
+ /**
+ * Called for each request after the corresponding {@link AccessExecutionGraph} has been retrieved from the
+ * {@link ExecutionGraphCache}.
+ *
+ * @param request for further information
+ * @param executionGraph for which the handler was called
+ * @return Response
+ * @throws RestHandlerException if the handler could not process the request
+ */
protected abstract R handleRequest(HandlerRequest<EmptyRequestBody, M> request, AccessExecutionGraph executionGraph) throws RestHandlerException;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/AbstractCheckpointHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/AbstractCheckpointHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/AbstractCheckpointHandler.java
index 62ed1a4..23d09f6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/AbstractCheckpointHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/AbstractCheckpointHandler.java
@@ -45,7 +45,7 @@ import java.util.concurrent.Executor;
*
* @param <R> type of the response
*/
-public abstract class AbstractCheckpointHandler<R extends ResponseBody> extends AbstractExecutionGraphHandler<R, CheckpointMessageParameters> {
+public abstract class AbstractCheckpointHandler<R extends ResponseBody, M extends CheckpointMessageParameters> extends AbstractExecutionGraphHandler<R, M> {
private final CheckpointStatsCache checkpointStatsCache;
@@ -53,7 +53,7 @@ public abstract class AbstractCheckpointHandler<R extends ResponseBody> extends
CompletableFuture<String> localRestAddress,
GatewayRetriever<? extends RestfulGateway> leaderRetriever,
Time timeout,
- MessageHeaders<EmptyRequestBody, R, CheckpointMessageParameters> messageHeaders,
+ MessageHeaders<EmptyRequestBody, R, M> messageHeaders,
ExecutionGraphCache executionGraphCache,
Executor executor,
CheckpointStatsCache checkpointStatsCache) {
@@ -63,7 +63,7 @@ public abstract class AbstractCheckpointHandler<R extends ResponseBody> extends
}
@Override
- protected R handleRequest(HandlerRequest<EmptyRequestBody, CheckpointMessageParameters> request, AccessExecutionGraph executionGraph) throws RestHandlerException {
+ protected R handleRequest(HandlerRequest<EmptyRequestBody, M> request, AccessExecutionGraph executionGraph) throws RestHandlerException {
final long checkpointId = request.getPathParameter(CheckpointIdPathParameter.class);
final CheckpointStatsSnapshot checkpointStatsSnapshot = executionGraph.getCheckpointStatsSnapshot();
@@ -78,7 +78,7 @@ public abstract class AbstractCheckpointHandler<R extends ResponseBody> extends
}
if (checkpointStats != null) {
- return handleCheckpointRequest(checkpointStats);
+ return handleCheckpointRequest(request, checkpointStats);
} else {
throw new RestHandlerException("Could not find checkpointing statistics for checkpoint " + checkpointId + '.', HttpResponseStatus.NOT_FOUND);
}
@@ -87,5 +87,13 @@ public abstract class AbstractCheckpointHandler<R extends ResponseBody> extends
}
}
- protected abstract R handleCheckpointRequest(AbstractCheckpointStats checkpointStats);
+ /**
+ * Called for each request with the corresponding {@link AbstractCheckpointStats} instance.
+ *
+ * @param request for further information
+ * @param checkpointStats for which the handler is called
+ * @return Response
+ * @throws RestHandlerException if the handler could not handle the request
+ */
+ protected abstract R handleCheckpointRequest(HandlerRequest<EmptyRequestBody, M> request, AbstractCheckpointStats checkpointStats) throws RestHandlerException;
}
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointStatisticDetailsHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointStatisticDetailsHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointStatisticDetailsHandler.java
index 2fc3008..e6b9f6b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointStatisticDetailsHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointStatisticDetailsHandler.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.rest.handler.job.checkpoints;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.runtime.checkpoint.AbstractCheckpointStats;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
import org.apache.flink.runtime.rest.handler.legacy.ExecutionGraphCache;
import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
import org.apache.flink.runtime.rest.messages.MessageHeaders;
@@ -34,7 +35,7 @@ import java.util.concurrent.Executor;
/**
* REST handler which returns the details for a checkpoint.
*/
-public class CheckpointStatisticDetailsHandler extends AbstractCheckpointHandler<CheckpointStatistics> {
+public class CheckpointStatisticDetailsHandler extends AbstractCheckpointHandler<CheckpointStatistics, CheckpointMessageParameters> {
public CheckpointStatisticDetailsHandler(
CompletableFuture<String> localRestAddress,
@@ -48,7 +49,7 @@ public class CheckpointStatisticDetailsHandler extends AbstractCheckpointHandler
}
@Override
- protected CheckpointStatistics handleCheckpointRequest(AbstractCheckpointStats checkpointStats) {
+ protected CheckpointStatistics handleCheckpointRequest(HandlerRequest<EmptyRequestBody, CheckpointMessageParameters> ignored, AbstractCheckpointStats checkpointStats) {
return CheckpointStatistics.generateCheckpointStatistics(checkpointStats, true);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointingStatisticsHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointingStatisticsHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointingStatisticsHandler.java
index 1c5762e..a8514d5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointingStatisticsHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/CheckpointingStatisticsHandler.java
@@ -36,6 +36,7 @@ import org.apache.flink.runtime.rest.messages.JobMessageParameters;
import org.apache.flink.runtime.rest.messages.MessageHeaders;
import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointStatistics;
import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointingStatistics;
+import org.apache.flink.runtime.rest.messages.checkpoints.MinMaxAvgStatistics;
import org.apache.flink.runtime.webmonitor.RestfulGateway;
import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
@@ -84,15 +85,15 @@ public class CheckpointingStatisticsHandler extends AbstractExecutionGraphHandle
final MinMaxAvgStats alignment = checkpointStatsSummary.getAlignmentBufferedStats();
final CheckpointingStatistics.Summary summary = new CheckpointingStatistics.Summary(
- new CheckpointingStatistics.MinMaxAvgStatistics(
+ new MinMaxAvgStatistics(
stateSize.getMinimum(),
stateSize.getMaximum(),
stateSize.getAverage()),
- new CheckpointingStatistics.MinMaxAvgStatistics(
+ new MinMaxAvgStatistics(
duration.getMinimum(),
duration.getMaximum(),
duration.getAverage()),
- new CheckpointingStatistics.MinMaxAvgStatistics(
+ new MinMaxAvgStatistics(
alignment.getMinimum(),
alignment.getMaximum(),
alignment.getAverage()));
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/TaskCheckpointStatisticDetailsHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/TaskCheckpointStatisticDetailsHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/TaskCheckpointStatisticDetailsHandler.java
new file mode 100644
index 0000000..5479159
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/checkpoints/TaskCheckpointStatisticDetailsHandler.java
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.handler.job.checkpoints;
+
+import org.apache.flink.api.common.time.Time;
+import org.apache.flink.runtime.checkpoint.AbstractCheckpointStats;
+import org.apache.flink.runtime.checkpoint.MinMaxAvgStats;
+import org.apache.flink.runtime.checkpoint.SubtaskStateStats;
+import org.apache.flink.runtime.checkpoint.TaskStateStats;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.rest.handler.HandlerRequest;
+import org.apache.flink.runtime.rest.handler.RestHandlerException;
+import org.apache.flink.runtime.rest.handler.legacy.ExecutionGraphCache;
+import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.JobVertexIdPathParameter;
+import org.apache.flink.runtime.rest.messages.MessageHeaders;
+import org.apache.flink.runtime.rest.messages.checkpoints.MinMaxAvgStatistics;
+import org.apache.flink.runtime.rest.messages.checkpoints.SubtaskCheckpointStatistics;
+import org.apache.flink.runtime.rest.messages.checkpoints.TaskCheckpointMessageParameters;
+import org.apache.flink.runtime.rest.messages.checkpoints.TaskCheckpointStatisticsWithSubtaskDetails;
+import org.apache.flink.runtime.webmonitor.RestfulGateway;
+import org.apache.flink.runtime.webmonitor.retriever.GatewayRetriever;
+
+import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+
+/**
+ * REST handler which serves checkpoint statistics for subtasks.
+ */
+public class TaskCheckpointStatisticDetailsHandler extends AbstractCheckpointHandler<TaskCheckpointStatisticsWithSubtaskDetails, TaskCheckpointMessageParameters> {
+
+ public TaskCheckpointStatisticDetailsHandler(
+ CompletableFuture<String> localRestAddress,
+ GatewayRetriever<? extends RestfulGateway> leaderRetriever,
+ Time timeout,
+ MessageHeaders<EmptyRequestBody, TaskCheckpointStatisticsWithSubtaskDetails, TaskCheckpointMessageParameters> messageHeaders,
+ ExecutionGraphCache executionGraphCache,
+ Executor executor,
+ CheckpointStatsCache checkpointStatsCache) {
+ super(localRestAddress, leaderRetriever, timeout, messageHeaders, executionGraphCache, executor, checkpointStatsCache);
+ }
+
+ @Override
+ protected TaskCheckpointStatisticsWithSubtaskDetails handleCheckpointRequest(
+ HandlerRequest<EmptyRequestBody, TaskCheckpointMessageParameters> request,
+ AbstractCheckpointStats checkpointStats) throws RestHandlerException {
+
+ final JobVertexID jobVertexId = request.getPathParameter(JobVertexIdPathParameter.class);
+
+ final TaskStateStats taskStatistics = checkpointStats.getTaskStateStats(jobVertexId);
+
+ if (taskStatistics != null) {
+
+ final TaskCheckpointStatisticsWithSubtaskDetails.Summary summary = createSummary(
+ taskStatistics.getSummaryStats(),
+ checkpointStats.getTriggerTimestamp());
+
+ final List<SubtaskCheckpointStatistics> subtaskCheckpointStatistics = createSubtaskCheckpointStatistics(
+ taskStatistics.getSubtaskStats(),
+ checkpointStats.getTriggerTimestamp());
+
+ return new TaskCheckpointStatisticsWithSubtaskDetails(
+ checkpointStats.getCheckpointId(),
+ checkpointStats.getStatus(),
+ taskStatistics.getLatestAckTimestamp(),
+ taskStatistics.getStateSize(),
+ taskStatistics.getEndToEndDuration(checkpointStats.getTriggerTimestamp()),
+ taskStatistics.getAlignmentBuffered(),
+ taskStatistics.getNumberOfSubtasks(),
+ taskStatistics.getNumberOfAcknowledgedSubtasks(),
+ summary,
+ subtaskCheckpointStatistics);
+ } else {
+ throw new RestHandlerException("There is no checkpoint statistics for task " + jobVertexId + '.', HttpResponseStatus.NOT_FOUND);
+ }
+ }
+
+ private static TaskCheckpointStatisticsWithSubtaskDetails.Summary createSummary(TaskStateStats.TaskStateStatsSummary taskStatisticsSummary, long triggerTimestamp) {
+ final MinMaxAvgStats stateSizeStats = taskStatisticsSummary.getStateSizeStats();
+ final MinMaxAvgStats ackTSStats = taskStatisticsSummary.getAckTimestampStats();
+ final MinMaxAvgStats syncDurationStats = taskStatisticsSummary.getSyncCheckpointDurationStats();
+ final MinMaxAvgStats asyncDurationStats = taskStatisticsSummary.getAsyncCheckpointDurationStats();
+
+ final TaskCheckpointStatisticsWithSubtaskDetails.CheckpointDuration checkpointDuration = new TaskCheckpointStatisticsWithSubtaskDetails.CheckpointDuration(
+ new MinMaxAvgStatistics(syncDurationStats.getMinimum(), syncDurationStats.getMaximum(), syncDurationStats.getAverage()),
+ new MinMaxAvgStatistics(asyncDurationStats.getMinimum(), asyncDurationStats.getMaximum(), asyncDurationStats.getAverage()));
+
+ final MinMaxAvgStats alignmentBufferedStats = taskStatisticsSummary.getAlignmentBufferedStats();
+ final MinMaxAvgStats alignmentDurationStats = taskStatisticsSummary.getAlignmentDurationStats();
+
+ final TaskCheckpointStatisticsWithSubtaskDetails.CheckpointAlignment checkpointAlignment = new TaskCheckpointStatisticsWithSubtaskDetails.CheckpointAlignment(
+ new MinMaxAvgStatistics(alignmentBufferedStats.getMinimum(), alignmentBufferedStats.getMaximum(), alignmentBufferedStats.getAverage()),
+ new MinMaxAvgStatistics(alignmentDurationStats.getMinimum(), alignmentDurationStats.getMaximum(), alignmentDurationStats.getAverage()));
+
+ return new TaskCheckpointStatisticsWithSubtaskDetails.Summary(
+ new MinMaxAvgStatistics(stateSizeStats.getMinimum(), stateSizeStats.getMaximum(), stateSizeStats.getAverage()),
+ new MinMaxAvgStatistics(
+ Math.max(0L, ackTSStats.getMinimum() - triggerTimestamp),
+ Math.max(0L, ackTSStats.getMaximum() - triggerTimestamp),
+ Math.max(0L, ackTSStats.getAverage() - triggerTimestamp)),
+ checkpointDuration,
+ checkpointAlignment);
+ }
+
+ private static List<SubtaskCheckpointStatistics> createSubtaskCheckpointStatistics(SubtaskStateStats[] subtaskStateStats, long triggerTimestamp) {
+ final List<SubtaskCheckpointStatistics> result = new ArrayList<>(subtaskStateStats.length);
+
+ for (int i = 0; i < subtaskStateStats.length; i++) {
+ final SubtaskStateStats subtask = subtaskStateStats[i];
+
+ if (subtask == null) {
+ result.add(new SubtaskCheckpointStatistics.PendingSubtaskCheckpointStatistics(i));
+ } else {
+ result.add(new SubtaskCheckpointStatistics.CompletedSubtaskCheckpointStatistics(
+ i,
+ subtask.getAckTimestamp(),
+ subtask.getEndToEndDuration(triggerTimestamp),
+ subtask.getStateSize(),
+ new SubtaskCheckpointStatistics.CompletedSubtaskCheckpointStatistics.CheckpointDuration(
+ subtask.getSyncCheckpointDuration(),
+ subtask.getAsyncCheckpointDuration()),
+ new SubtaskCheckpointStatistics.CompletedSubtaskCheckpointStatistics.CheckpointAlignment(
+ subtask.getAlignmentBuffered(),
+ subtask.getAlignmentDuration())
+ ));
+ }
+ }
+
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/ClusterConfigHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/ClusterConfigHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/ClusterConfigHandler.java
index d3ea160..1de5eca 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/ClusterConfigHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/ClusterConfigHandler.java
@@ -23,8 +23,8 @@ import org.apache.flink.runtime.dispatcher.DispatcherGateway;
import org.apache.flink.runtime.jobmaster.JobManagerGateway;
import org.apache.flink.runtime.rest.handler.HandlerRequest;
import org.apache.flink.runtime.rest.handler.LegacyRestHandler;
-import org.apache.flink.runtime.rest.handler.legacy.messages.ClusterConfigurationInfo;
-import org.apache.flink.runtime.rest.handler.legacy.messages.ClusterConfigurationInfoEntry;
+import org.apache.flink.runtime.rest.messages.ClusterConfigurationInfo;
+import org.apache.flink.runtime.rest.messages.ClusterConfigurationInfoEntry;
import org.apache.flink.runtime.rest.messages.ClusterConfigurationInfoHeaders;
import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/ClusterOverviewHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/ClusterOverviewHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/ClusterOverviewHandler.java
index 2f883ab..ff05088 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/ClusterOverviewHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/ClusterOverviewHandler.java
@@ -26,9 +26,9 @@ import org.apache.flink.runtime.messages.webmonitor.JobsOverview;
import org.apache.flink.runtime.messages.webmonitor.StatusOverview;
import org.apache.flink.runtime.rest.handler.HandlerRequest;
import org.apache.flink.runtime.rest.handler.LegacyRestHandler;
-import org.apache.flink.runtime.rest.handler.legacy.messages.StatusOverviewWithVersion;
import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
+import org.apache.flink.runtime.rest.messages.StatusOverviewWithVersion;
import org.apache.flink.runtime.util.EnvironmentInformation;
import org.apache.flink.util.FlinkException;
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/DashboardConfigHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/DashboardConfigHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/DashboardConfigHandler.java
index 0cef5fb..b5aa036 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/DashboardConfigHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/DashboardConfigHandler.java
@@ -22,7 +22,7 @@ import org.apache.flink.runtime.dispatcher.DispatcherGateway;
import org.apache.flink.runtime.jobmaster.JobManagerGateway;
import org.apache.flink.runtime.rest.handler.HandlerRequest;
import org.apache.flink.runtime.rest.handler.LegacyRestHandler;
-import org.apache.flink.runtime.rest.handler.legacy.messages.DashboardConfiguration;
+import org.apache.flink.runtime.rest.messages.DashboardConfiguration;
import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/checkpoints/CheckpointStatsHandler.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/checkpoints/CheckpointStatsHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/checkpoints/CheckpointStatsHandler.java
index b6c86be..38cd0c8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/checkpoints/CheckpointStatsHandler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/checkpoints/CheckpointStatsHandler.java
@@ -33,6 +33,7 @@ import org.apache.flink.runtime.rest.handler.legacy.ExecutionGraphCache;
import org.apache.flink.runtime.rest.handler.legacy.JsonFactory;
import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointStatistics;
import org.apache.flink.runtime.rest.messages.checkpoints.CheckpointingStatistics;
+import org.apache.flink.runtime.rest.messages.checkpoints.MinMaxAvgStatistics;
import org.apache.flink.runtime.webmonitor.history.ArchivedJson;
import org.apache.flink.runtime.webmonitor.history.JsonArchivist;
import org.apache.flink.util.FlinkException;
@@ -158,9 +159,9 @@ public class CheckpointStatsHandler extends AbstractExecutionGraphRequestHandler
}
static void writeMinMaxAvg(JsonGenerator gen, MinMaxAvgStats minMaxAvg) throws IOException {
- gen.writeNumberField(CheckpointingStatistics.MinMaxAvgStatistics.FIELD_NAME_MINIMUM, minMaxAvg.getMinimum());
- gen.writeNumberField(CheckpointingStatistics.MinMaxAvgStatistics.FIELD_NAME_MAXIMUM, minMaxAvg.getMaximum());
- gen.writeNumberField(CheckpointingStatistics.MinMaxAvgStatistics.FIELD_NAME_AVERAGE, minMaxAvg.getAverage());
+ gen.writeNumberField(MinMaxAvgStatistics.FIELD_NAME_MINIMUM, minMaxAvg.getMinimum());
+ gen.writeNumberField(MinMaxAvgStatistics.FIELD_NAME_MAXIMUM, minMaxAvg.getMaximum());
+ gen.writeNumberField(MinMaxAvgStatistics.FIELD_NAME_AVERAGE, minMaxAvg.getAverage());
}
private static void writeLatestCheckpoints(
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfo.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfo.java
deleted file mode 100644
index bf527d9..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfo.java
+++ /dev/null
@@ -1,59 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.rest.handler.legacy.messages;
-
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.rest.handler.legacy.ClusterConfigHandler;
-import org.apache.flink.runtime.rest.messages.ResponseBody;
-
-import java.util.ArrayList;
-
-/**
- * Response of the {@link ClusterConfigHandler}, respresented as a list
- * of key-value pairs of the cluster {@link Configuration}.
- */
-public class ClusterConfigurationInfo extends ArrayList<ClusterConfigurationInfoEntry> implements ResponseBody {
-
- private static final long serialVersionUID = -1170348873871206964L;
-
- // a default constructor is required for collection type marshalling
- public ClusterConfigurationInfo() {}
-
- public ClusterConfigurationInfo(int initialEntries) {
- super(initialEntries);
- }
-
- public static ClusterConfigurationInfo from(Configuration config) {
- ClusterConfigurationInfo clusterConfig = new ClusterConfigurationInfo(config.keySet().size());
-
- for (String key : config.keySet()) {
- String value = config.getString(key, null);
-
- // Mask key values which contain sensitive information
- if (value != null && key.toLowerCase().contains("password")) {
- value = "******";
- }
-
- clusterConfig.add(new ClusterConfigurationInfoEntry(key, value));
- }
-
- return clusterConfig;
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfoEntry.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfoEntry.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfoEntry.java
deleted file mode 100644
index 19e3821..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/ClusterConfigurationInfoEntry.java
+++ /dev/null
@@ -1,81 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.rest.handler.legacy.messages;
-
-import org.apache.flink.util.Preconditions;
-
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
-
-import java.util.Objects;
-
-/**
- * A single key-value pair entry in the {@link ClusterConfigurationInfo} response.
- */
-public class ClusterConfigurationInfoEntry {
-
- public static final String FIELD_NAME_CONFIG_KEY = "key";
- public static final String FIELD_NAME_CONFIG_VALUE = "value";
-
- @JsonProperty(FIELD_NAME_CONFIG_KEY)
- private final String key;
-
- @JsonProperty(FIELD_NAME_CONFIG_VALUE)
- private final String value;
-
- @JsonCreator
- public ClusterConfigurationInfoEntry(
- @JsonProperty(FIELD_NAME_CONFIG_KEY) String key,
- @JsonProperty(FIELD_NAME_CONFIG_VALUE) String value) {
- this.key = Preconditions.checkNotNull(key);
- this.value = Preconditions.checkNotNull(value);
- }
-
- public String getKey() {
- return key;
- }
-
- public String getValue() {
- return value;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
-
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
-
- ClusterConfigurationInfoEntry that = (ClusterConfigurationInfoEntry) o;
- return key.equals(that.key) && value.equals(that.value);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(key, value);
- }
-
- @Override
- public String toString() {
- return "(" + key + "," + value + ")";
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/DashboardConfiguration.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/DashboardConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/DashboardConfiguration.java
deleted file mode 100644
index cfb3aaa..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/DashboardConfiguration.java
+++ /dev/null
@@ -1,137 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.rest.handler.legacy.messages;
-
-import org.apache.flink.runtime.rest.handler.legacy.DashboardConfigHandler;
-import org.apache.flink.runtime.rest.messages.ResponseBody;
-import org.apache.flink.runtime.util.EnvironmentInformation;
-import org.apache.flink.util.Preconditions;
-
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
-
-import java.time.ZonedDateTime;
-import java.time.format.TextStyle;
-import java.util.Locale;
-import java.util.Objects;
-
-/**
- * Response of the {@link DashboardConfigHandler} containing general configuration
- * values such as the time zone and the refresh interval.
- */
-public class DashboardConfiguration implements ResponseBody {
-
- public static final String FIELD_NAME_REFRESH_INTERVAL = "refresh-interval";
- public static final String FIELD_NAME_TIMEZONE_OFFSET = "timezone-offset";
- public static final String FIELD_NAME_TIMEZONE_NAME = "timezone-name";
- public static final String FIELD_NAME_FLINK_VERSION = "flink-version";
- public static final String FIELD_NAME_FLINK_REVISION = "flink-revision";
-
- @JsonProperty(FIELD_NAME_REFRESH_INTERVAL)
- private final long refreshInterval;
-
- @JsonProperty(FIELD_NAME_TIMEZONE_NAME)
- private final String timeZoneName;
-
- @JsonProperty(FIELD_NAME_TIMEZONE_OFFSET)
- private final int timeZoneOffset;
-
- @JsonProperty(FIELD_NAME_FLINK_VERSION)
- private final String flinkVersion;
-
- @JsonProperty(FIELD_NAME_FLINK_REVISION)
- private final String flinkRevision;
-
- @JsonCreator
- public DashboardConfiguration(
- @JsonProperty(FIELD_NAME_REFRESH_INTERVAL) long refreshInterval,
- @JsonProperty(FIELD_NAME_TIMEZONE_NAME) String timeZoneName,
- @JsonProperty(FIELD_NAME_TIMEZONE_OFFSET) int timeZoneOffset,
- @JsonProperty(FIELD_NAME_FLINK_VERSION) String flinkVersion,
- @JsonProperty(FIELD_NAME_FLINK_REVISION) String flinkRevision) {
- this.refreshInterval = refreshInterval;
- this.timeZoneName = Preconditions.checkNotNull(timeZoneName);
- this.timeZoneOffset = timeZoneOffset;
- this.flinkVersion = Preconditions.checkNotNull(flinkVersion);
- this.flinkRevision = Preconditions.checkNotNull(flinkRevision);
- }
-
- public long getRefreshInterval() {
- return refreshInterval;
- }
-
- public int getTimeZoneOffset() {
- return timeZoneOffset;
- }
-
- public String getTimeZoneName() {
- return timeZoneName;
- }
-
- public String getFlinkVersion() {
- return flinkVersion;
- }
-
- public String getFlinkRevision() {
- return flinkRevision;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- DashboardConfiguration that = (DashboardConfiguration) o;
- return refreshInterval == that.refreshInterval &&
- timeZoneOffset == that.timeZoneOffset &&
- Objects.equals(timeZoneName, that.timeZoneName) &&
- Objects.equals(flinkVersion, that.flinkVersion) &&
- Objects.equals(flinkRevision, that.flinkRevision);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(refreshInterval, timeZoneName, timeZoneOffset, flinkVersion, flinkRevision);
- }
-
- public static DashboardConfiguration from(long refreshInterval, ZonedDateTime zonedDateTime) {
-
- final String flinkVersion = EnvironmentInformation.getVersion();
-
- final EnvironmentInformation.RevisionInformation revision = EnvironmentInformation.getRevisionInformation();
- final String flinkRevision;
-
- if (revision != null) {
- flinkRevision = revision.commitId + " @ " + revision.commitDate;
- } else {
- flinkRevision = "unknown revision";
- }
-
- return new DashboardConfiguration(
- refreshInterval,
- zonedDateTime.getZone().getDisplayName(TextStyle.FULL, Locale.getDefault()),
- // convert zone date time into offset in order to not do the day light saving adaptions wrt the offset
- zonedDateTime.toOffsetDateTime().getOffset().getTotalSeconds() * 1000,
- flinkVersion,
- flinkRevision);
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/StatusOverviewWithVersion.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/StatusOverviewWithVersion.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/StatusOverviewWithVersion.java
deleted file mode 100644
index f001afc..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/legacy/messages/StatusOverviewWithVersion.java
+++ /dev/null
@@ -1,130 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.runtime.rest.handler.legacy.messages;
-
-import org.apache.flink.runtime.messages.webmonitor.JobsOverview;
-import org.apache.flink.runtime.messages.webmonitor.StatusOverview;
-import org.apache.flink.runtime.rest.messages.ResponseBody;
-import org.apache.flink.util.Preconditions;
-
-import com.fasterxml.jackson.annotation.JsonCreator;
-import com.fasterxml.jackson.annotation.JsonProperty;
-
-import java.util.Objects;
-
-/**
- * Status overview message including the current Flink version and commit id.
- */
-public class StatusOverviewWithVersion extends StatusOverview implements ResponseBody {
-
- private static final long serialVersionUID = 5000058311783413216L;
-
- public static final String FIELD_NAME_VERSION = "flink-version";
- public static final String FIELD_NAME_COMMIT = "flink-commit";
-
- @JsonProperty(FIELD_NAME_VERSION)
- private final String version;
-
- @JsonProperty(FIELD_NAME_COMMIT)
- private final String commitId;
-
- @JsonCreator
- public StatusOverviewWithVersion(
- @JsonProperty(FIELD_NAME_TASKMANAGERS) int numTaskManagersConnected,
- @JsonProperty(FIELD_NAME_SLOTS_TOTAL) int numSlotsTotal,
- @JsonProperty(FIELD_NAME_SLOTS_AVAILABLE) int numSlotsAvailable,
- @JsonProperty(FIELD_NAME_JOBS_RUNNING) int numJobsRunningOrPending,
- @JsonProperty(FIELD_NAME_JOBS_FINISHED) int numJobsFinished,
- @JsonProperty(FIELD_NAME_JOBS_CANCELLED) int numJobsCancelled,
- @JsonProperty(FIELD_NAME_JOBS_FAILED) int numJobsFailed,
- @JsonProperty(FIELD_NAME_VERSION) String version,
- @JsonProperty(FIELD_NAME_COMMIT) String commitId) {
- super(
- numTaskManagersConnected,
- numSlotsTotal,
- numSlotsAvailable,
- numJobsRunningOrPending,
- numJobsFinished,
- numJobsCancelled,
- numJobsFailed);
-
- this.version = Preconditions.checkNotNull(version);
- this.commitId = Preconditions.checkNotNull(commitId);
- }
-
- public StatusOverviewWithVersion(
- int numTaskManagersConnected,
- int numSlotsTotal,
- int numSlotsAvailable,
- JobsOverview jobs1,
- JobsOverview jobs2,
- String version,
- String commitId) {
- super(numTaskManagersConnected, numSlotsTotal, numSlotsAvailable, jobs1, jobs2);
-
- this.version = Preconditions.checkNotNull(version);
- this.commitId = Preconditions.checkNotNull(commitId);
- }
-
- public static StatusOverviewWithVersion fromStatusOverview(StatusOverview statusOverview, String version, String commitId) {
- return new StatusOverviewWithVersion(
- statusOverview.getNumTaskManagersConnected(),
- statusOverview.getNumSlotsTotal(),
- statusOverview.getNumSlotsAvailable(),
- statusOverview.getNumJobsRunningOrPending(),
- statusOverview.getNumJobsFinished(),
- statusOverview.getNumJobsCancelled(),
- statusOverview.getNumJobsFailed(),
- version,
- commitId);
- }
-
- public String getVersion() {
- return version;
- }
-
- public String getCommitId() {
- return commitId;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- if (!super.equals(o)) {
- return false;
- }
-
- StatusOverviewWithVersion that = (StatusOverviewWithVersion) o;
-
- return Objects.equals(version, that.getVersion()) && Objects.equals(commitId, that.getCommitId());
- }
-
- @Override
- public int hashCode() {
- int result = super.hashCode();
- result = 31 * result + (version != null ? version.hashCode() : 0);
- result = 31 * result + (commitId != null ? commitId.hashCode() : 0);
- return result;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfo.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfo.java
new file mode 100644
index 0000000..dba6a4c
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfo.java
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.rest.handler.legacy.ClusterConfigHandler;
+
+import java.util.ArrayList;
+
+/**
+ * Response of the {@link ClusterConfigHandler}, respresented as a list
+ * of key-value pairs of the cluster {@link Configuration}.
+ */
+public class ClusterConfigurationInfo extends ArrayList<ClusterConfigurationInfoEntry> implements ResponseBody {
+
+ private static final long serialVersionUID = -1170348873871206964L;
+
+ // a default constructor is required for collection type marshalling
+ public ClusterConfigurationInfo() {}
+
+ public ClusterConfigurationInfo(int initialEntries) {
+ super(initialEntries);
+ }
+
+ public static ClusterConfigurationInfo from(Configuration config) {
+ ClusterConfigurationInfo clusterConfig = new ClusterConfigurationInfo(config.keySet().size());
+
+ for (String key : config.keySet()) {
+ String value = config.getString(key, null);
+
+ // Mask key values which contain sensitive information
+ if (value != null && key.toLowerCase().contains("password")) {
+ value = "******";
+ }
+
+ clusterConfig.add(new ClusterConfigurationInfoEntry(key, value));
+ }
+
+ return clusterConfig;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfoEntry.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfoEntry.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfoEntry.java
new file mode 100644
index 0000000..634f728
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfoEntry.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages;
+
+import org.apache.flink.util.Preconditions;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Objects;
+
+/**
+ * A single key-value pair entry in the {@link ClusterConfigurationInfo} response.
+ */
+public class ClusterConfigurationInfoEntry {
+
+ public static final String FIELD_NAME_CONFIG_KEY = "key";
+ public static final String FIELD_NAME_CONFIG_VALUE = "value";
+
+ @JsonProperty(FIELD_NAME_CONFIG_KEY)
+ private final String key;
+
+ @JsonProperty(FIELD_NAME_CONFIG_VALUE)
+ private final String value;
+
+ @JsonCreator
+ public ClusterConfigurationInfoEntry(
+ @JsonProperty(FIELD_NAME_CONFIG_KEY) String key,
+ @JsonProperty(FIELD_NAME_CONFIG_VALUE) String value) {
+ this.key = Preconditions.checkNotNull(key);
+ this.value = Preconditions.checkNotNull(value);
+ }
+
+ public String getKey() {
+ return key;
+ }
+
+ public String getValue() {
+ return value;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ ClusterConfigurationInfoEntry that = (ClusterConfigurationInfoEntry) o;
+ return key.equals(that.key) && value.equals(that.value);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(key, value);
+ }
+
+ @Override
+ public String toString() {
+ return "(" + key + "," + value + ")";
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfoHeaders.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfoHeaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfoHeaders.java
index 0ad56b1..11a750a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfoHeaders.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterConfigurationInfoHeaders.java
@@ -20,7 +20,6 @@ package org.apache.flink.runtime.rest.messages;
import org.apache.flink.runtime.rest.HttpMethodWrapper;
import org.apache.flink.runtime.rest.handler.legacy.ClusterConfigHandler;
-import org.apache.flink.runtime.rest.handler.legacy.messages.ClusterConfigurationInfo;
import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterOverviewHeaders.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterOverviewHeaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterOverviewHeaders.java
index 887ce2e..f8411e4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterOverviewHeaders.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/ClusterOverviewHeaders.java
@@ -20,7 +20,6 @@ package org.apache.flink.runtime.rest.messages;
import org.apache.flink.runtime.rest.HttpMethodWrapper;
import org.apache.flink.runtime.rest.handler.legacy.ClusterOverviewHandler;
-import org.apache.flink.runtime.rest.handler.legacy.messages.StatusOverviewWithVersion;
import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/DashboardConfiguration.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/DashboardConfiguration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/DashboardConfiguration.java
new file mode 100644
index 0000000..3e9af68
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/DashboardConfiguration.java
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages;
+
+import org.apache.flink.runtime.rest.handler.legacy.DashboardConfigHandler;
+import org.apache.flink.runtime.util.EnvironmentInformation;
+import org.apache.flink.util.Preconditions;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.time.ZonedDateTime;
+import java.time.format.TextStyle;
+import java.util.Locale;
+import java.util.Objects;
+
+/**
+ * Response of the {@link DashboardConfigHandler} containing general configuration
+ * values such as the time zone and the refresh interval.
+ */
+public class DashboardConfiguration implements ResponseBody {
+
+ public static final String FIELD_NAME_REFRESH_INTERVAL = "refresh-interval";
+ public static final String FIELD_NAME_TIMEZONE_OFFSET = "timezone-offset";
+ public static final String FIELD_NAME_TIMEZONE_NAME = "timezone-name";
+ public static final String FIELD_NAME_FLINK_VERSION = "flink-version";
+ public static final String FIELD_NAME_FLINK_REVISION = "flink-revision";
+
+ @JsonProperty(FIELD_NAME_REFRESH_INTERVAL)
+ private final long refreshInterval;
+
+ @JsonProperty(FIELD_NAME_TIMEZONE_NAME)
+ private final String timeZoneName;
+
+ @JsonProperty(FIELD_NAME_TIMEZONE_OFFSET)
+ private final int timeZoneOffset;
+
+ @JsonProperty(FIELD_NAME_FLINK_VERSION)
+ private final String flinkVersion;
+
+ @JsonProperty(FIELD_NAME_FLINK_REVISION)
+ private final String flinkRevision;
+
+ @JsonCreator
+ public DashboardConfiguration(
+ @JsonProperty(FIELD_NAME_REFRESH_INTERVAL) long refreshInterval,
+ @JsonProperty(FIELD_NAME_TIMEZONE_NAME) String timeZoneName,
+ @JsonProperty(FIELD_NAME_TIMEZONE_OFFSET) int timeZoneOffset,
+ @JsonProperty(FIELD_NAME_FLINK_VERSION) String flinkVersion,
+ @JsonProperty(FIELD_NAME_FLINK_REVISION) String flinkRevision) {
+ this.refreshInterval = refreshInterval;
+ this.timeZoneName = Preconditions.checkNotNull(timeZoneName);
+ this.timeZoneOffset = timeZoneOffset;
+ this.flinkVersion = Preconditions.checkNotNull(flinkVersion);
+ this.flinkRevision = Preconditions.checkNotNull(flinkRevision);
+ }
+
+ public long getRefreshInterval() {
+ return refreshInterval;
+ }
+
+ public int getTimeZoneOffset() {
+ return timeZoneOffset;
+ }
+
+ public String getTimeZoneName() {
+ return timeZoneName;
+ }
+
+ public String getFlinkVersion() {
+ return flinkVersion;
+ }
+
+ public String getFlinkRevision() {
+ return flinkRevision;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ DashboardConfiguration that = (DashboardConfiguration) o;
+ return refreshInterval == that.refreshInterval &&
+ timeZoneOffset == that.timeZoneOffset &&
+ Objects.equals(timeZoneName, that.timeZoneName) &&
+ Objects.equals(flinkVersion, that.flinkVersion) &&
+ Objects.equals(flinkRevision, that.flinkRevision);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(refreshInterval, timeZoneName, timeZoneOffset, flinkVersion, flinkRevision);
+ }
+
+ public static DashboardConfiguration from(long refreshInterval, ZonedDateTime zonedDateTime) {
+
+ final String flinkVersion = EnvironmentInformation.getVersion();
+
+ final EnvironmentInformation.RevisionInformation revision = EnvironmentInformation.getRevisionInformation();
+ final String flinkRevision;
+
+ if (revision != null) {
+ flinkRevision = revision.commitId + " @ " + revision.commitDate;
+ } else {
+ flinkRevision = "unknown revision";
+ }
+
+ return new DashboardConfiguration(
+ refreshInterval,
+ zonedDateTime.getZone().getDisplayName(TextStyle.FULL, Locale.getDefault()),
+ // convert zone date time into offset in order to not do the day light saving adaptions wrt the offset
+ zonedDateTime.toOffsetDateTime().getOffset().getTotalSeconds() * 1000,
+ flinkVersion,
+ flinkRevision);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/DashboardConfigurationHeaders.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/DashboardConfigurationHeaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/DashboardConfigurationHeaders.java
index cc03b7b..cb14fc5 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/DashboardConfigurationHeaders.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/DashboardConfigurationHeaders.java
@@ -20,7 +20,6 @@ package org.apache.flink.runtime.rest.messages;
import org.apache.flink.runtime.rest.HttpMethodWrapper;
import org.apache.flink.runtime.rest.handler.legacy.DashboardConfigHandler;
-import org.apache.flink.runtime.rest.handler.legacy.messages.DashboardConfiguration;
import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobVertexIdPathParameter.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobVertexIdPathParameter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobVertexIdPathParameter.java
new file mode 100644
index 0000000..0e000f6
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/JobVertexIdPathParameter.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages;
+
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+
+/**
+ * Path parameter specifying a {@link JobVertexID}.
+ */
+public class JobVertexIdPathParameter extends MessagePathParameter<JobVertexID> {
+
+ private static final String KEY = "vertexid";
+
+ public JobVertexIdPathParameter() {
+ super(KEY);
+ }
+
+ @Override
+ protected JobVertexID convertFromString(String value) throws ConversionException {
+ return JobVertexID.fromHexString(value);
+ }
+
+ @Override
+ protected String convertToString(JobVertexID value) {
+ return value.toString();
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/StatusOverviewWithVersion.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/StatusOverviewWithVersion.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/StatusOverviewWithVersion.java
new file mode 100644
index 0000000..1ee5c86
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/StatusOverviewWithVersion.java
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages;
+
+import org.apache.flink.runtime.messages.webmonitor.JobsOverview;
+import org.apache.flink.runtime.messages.webmonitor.StatusOverview;
+import org.apache.flink.util.Preconditions;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Objects;
+
+/**
+ * Status overview message including the current Flink version and commit id.
+ */
+public class StatusOverviewWithVersion extends StatusOverview implements ResponseBody {
+
+ private static final long serialVersionUID = 5000058311783413216L;
+
+ public static final String FIELD_NAME_VERSION = "flink-version";
+ public static final String FIELD_NAME_COMMIT = "flink-commit";
+
+ @JsonProperty(FIELD_NAME_VERSION)
+ private final String version;
+
+ @JsonProperty(FIELD_NAME_COMMIT)
+ private final String commitId;
+
+ @JsonCreator
+ public StatusOverviewWithVersion(
+ @JsonProperty(FIELD_NAME_TASKMANAGERS) int numTaskManagersConnected,
+ @JsonProperty(FIELD_NAME_SLOTS_TOTAL) int numSlotsTotal,
+ @JsonProperty(FIELD_NAME_SLOTS_AVAILABLE) int numSlotsAvailable,
+ @JsonProperty(FIELD_NAME_JOBS_RUNNING) int numJobsRunningOrPending,
+ @JsonProperty(FIELD_NAME_JOBS_FINISHED) int numJobsFinished,
+ @JsonProperty(FIELD_NAME_JOBS_CANCELLED) int numJobsCancelled,
+ @JsonProperty(FIELD_NAME_JOBS_FAILED) int numJobsFailed,
+ @JsonProperty(FIELD_NAME_VERSION) String version,
+ @JsonProperty(FIELD_NAME_COMMIT) String commitId) {
+ super(
+ numTaskManagersConnected,
+ numSlotsTotal,
+ numSlotsAvailable,
+ numJobsRunningOrPending,
+ numJobsFinished,
+ numJobsCancelled,
+ numJobsFailed);
+
+ this.version = Preconditions.checkNotNull(version);
+ this.commitId = Preconditions.checkNotNull(commitId);
+ }
+
+ public StatusOverviewWithVersion(
+ int numTaskManagersConnected,
+ int numSlotsTotal,
+ int numSlotsAvailable,
+ JobsOverview jobs1,
+ JobsOverview jobs2,
+ String version,
+ String commitId) {
+ super(numTaskManagersConnected, numSlotsTotal, numSlotsAvailable, jobs1, jobs2);
+
+ this.version = Preconditions.checkNotNull(version);
+ this.commitId = Preconditions.checkNotNull(commitId);
+ }
+
+ public static StatusOverviewWithVersion fromStatusOverview(StatusOverview statusOverview, String version, String commitId) {
+ return new StatusOverviewWithVersion(
+ statusOverview.getNumTaskManagersConnected(),
+ statusOverview.getNumSlotsTotal(),
+ statusOverview.getNumSlotsAvailable(),
+ statusOverview.getNumJobsRunningOrPending(),
+ statusOverview.getNumJobsFinished(),
+ statusOverview.getNumJobsCancelled(),
+ statusOverview.getNumJobsFailed(),
+ version,
+ commitId);
+ }
+
+ public String getVersion() {
+ return version;
+ }
+
+ public String getCommitId() {
+ return commitId;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+
+ StatusOverviewWithVersion that = (StatusOverviewWithVersion) o;
+
+ return Objects.equals(version, that.getVersion()) && Objects.equals(commitId, that.getCommitId());
+ }
+
+ @Override
+ public int hashCode() {
+ int result = super.hashCode();
+ result = 31 * result + (version != null ? version.hashCode() : 0);
+ result = 31 * result + (commitId != null ? commitId.hashCode() : 0);
+ return result;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointStatistics.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointStatistics.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointStatistics.java
index 9fb1094..4144571 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointStatistics.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointStatistics.java
@@ -225,6 +225,8 @@ public class CheckpointStatistics implements ResponseBody {
checkpointStatisticsPerTask.put(
taskStateStat.getJobVertexId(),
new TaskCheckpointStatistics(
+ checkpointStats.getCheckpointId(),
+ checkpointStats.getStatus(),
taskStateStat.getLatestAckTimestamp(),
taskStateStat.getStateSize(),
taskStateStat.getEndToEndDuration(checkpointStats.getTriggerTimestamp()),
@@ -280,104 +282,6 @@ public class CheckpointStatistics implements ResponseBody {
// ---------------------------------------------------------------------
/**
- * Checkpoint statistics for a single task.
- */
- public static final class TaskCheckpointStatistics {
-
- public static final String FIELD_NAME_LATEST_ACK_TIMESTAMP = "latest_ack_timestamp";
-
- public static final String FIELD_NAME_STATE_SIZE = "state_size";
-
- public static final String FIELD_NAME_DURATION = "end_to_end_duration";
-
- public static final String FIELD_NAME_ALIGNMENT_BUFFERED = "alignment_buffered";
-
- public static final String FIELD_NAME_NUM_SUBTASKS = "num_subtasks";
-
- public static final String FIELD_NAME_NUM_ACK_SUBTASKS = "num_acknowledged_subtasks";
-
- @JsonProperty(FIELD_NAME_LATEST_ACK_TIMESTAMP)
- private final long latestAckTimestamp;
-
- @JsonProperty(FIELD_NAME_STATE_SIZE)
- private final long stateSize;
-
- @JsonProperty(FIELD_NAME_DURATION)
- private final long duration;
-
- @JsonProperty(FIELD_NAME_ALIGNMENT_BUFFERED)
- private final long alignmentBuffered;
-
- @JsonProperty(FIELD_NAME_NUM_SUBTASKS)
- private final int numSubtasks;
-
- @JsonProperty(FIELD_NAME_NUM_ACK_SUBTASKS)
- private final int numAckSubtasks;
-
- @JsonCreator
- public TaskCheckpointStatistics(
- @JsonProperty(FIELD_NAME_LATEST_ACK_TIMESTAMP) long latestAckTimestamp,
- @JsonProperty(FIELD_NAME_STATE_SIZE) long stateSize,
- @JsonProperty(FIELD_NAME_DURATION) long duration,
- @JsonProperty(FIELD_NAME_ALIGNMENT_BUFFERED) long alignmentBuffered,
- @JsonProperty(FIELD_NAME_NUM_SUBTASKS) int numSubtasks,
- @JsonProperty(FIELD_NAME_NUM_ACK_SUBTASKS) int numAckSubtasks) {
- this.latestAckTimestamp = latestAckTimestamp;
- this.stateSize = stateSize;
- this.duration = duration;
- this.alignmentBuffered = alignmentBuffered;
- this.numSubtasks = numSubtasks;
- this.numAckSubtasks = numAckSubtasks;
- }
-
- public long getLatestAckTimestamp() {
- return latestAckTimestamp;
- }
-
- public long getStateSize() {
- return stateSize;
- }
-
- public long getDuration() {
- return duration;
- }
-
- public long getAlignmentBuffered() {
- return alignmentBuffered;
- }
-
- public int getNumSubtasks() {
- return numSubtasks;
- }
-
- public int getNumAckSubtasks() {
- return numAckSubtasks;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- TaskCheckpointStatistics that = (TaskCheckpointStatistics) o;
- return latestAckTimestamp == that.latestAckTimestamp &&
- stateSize == that.stateSize &&
- duration == that.duration &&
- alignmentBuffered == that.alignmentBuffered &&
- numSubtasks == that.numSubtasks &&
- numAckSubtasks == that.numAckSubtasks;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(latestAckTimestamp, stateSize, duration, alignmentBuffered, numSubtasks, numAckSubtasks);
- }
- }
-
- /**
* Statistics for a completed checkpoint.
*/
public static final class CompletedCheckpointStatistics extends CheckpointStatistics {
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointingStatistics.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointingStatistics.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointingStatistics.java
index 1f00fcc..96ca227 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointingStatistics.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/CheckpointingStatistics.java
@@ -257,68 +257,6 @@ public class CheckpointingStatistics implements ResponseBody {
}
/**
- * Minimum, maximum and average statistics.
- */
- public static final class MinMaxAvgStatistics {
-
- public static final String FIELD_NAME_MINIMUM = "min";
-
- public static final String FIELD_NAME_MAXIMUM = "max";
-
- public static final String FIELD_NAME_AVERAGE = "avg";
-
- @JsonProperty(FIELD_NAME_MINIMUM)
- private final long minimum;
-
- @JsonProperty(FIELD_NAME_MAXIMUM)
- private final long maximum;
-
- @JsonProperty(FIELD_NAME_AVERAGE)
- private final long average;
-
- @JsonCreator
- public MinMaxAvgStatistics(
- @JsonProperty(FIELD_NAME_MINIMUM) long minimum,
- @JsonProperty(FIELD_NAME_MAXIMUM) long maximum,
- @JsonProperty(FIELD_NAME_AVERAGE) long average) {
- this.minimum = minimum;
- this.maximum = maximum;
- this.average = average;
- }
-
- public long getMinimum() {
- return minimum;
- }
-
- public long getMaximum() {
- return maximum;
- }
-
- public long getAverage() {
- return average;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- MinMaxAvgStatistics that = (MinMaxAvgStatistics) o;
- return minimum == that.minimum &&
- maximum == that.maximum &&
- average == that.average;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(minimum, maximum, average);
- }
- }
-
- /**
* Statistics about the latest checkpoints.
*/
public static final class LatestCheckpoints {
http://git-wip-us.apache.org/repos/asf/flink/blob/bc4638a3/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/MinMaxAvgStatistics.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/MinMaxAvgStatistics.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/MinMaxAvgStatistics.java
new file mode 100644
index 0000000..171dbac
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/checkpoints/MinMaxAvgStatistics.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.rest.messages.checkpoints;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+import java.util.Objects;
+
+/**
+ * Minimum, maximum and average statistics.
+ */
+public final class MinMaxAvgStatistics {
+
+ public static final String FIELD_NAME_MINIMUM = "min";
+
+ public static final String FIELD_NAME_MAXIMUM = "max";
+
+ public static final String FIELD_NAME_AVERAGE = "avg";
+
+ @JsonProperty(FIELD_NAME_MINIMUM)
+ private final long minimum;
+
+ @JsonProperty(FIELD_NAME_MAXIMUM)
+ private final long maximum;
+
+ @JsonProperty(FIELD_NAME_AVERAGE)
+ private final long average;
+
+ @JsonCreator
+ public MinMaxAvgStatistics(
+ @JsonProperty(FIELD_NAME_MINIMUM) long minimum,
+ @JsonProperty(FIELD_NAME_MAXIMUM) long maximum,
+ @JsonProperty(FIELD_NAME_AVERAGE) long average) {
+ this.minimum = minimum;
+ this.maximum = maximum;
+ this.average = average;
+ }
+
+ public long getMinimum() {
+ return minimum;
+ }
+
+ public long getMaximum() {
+ return maximum;
+ }
+
+ public long getAverage() {
+ return average;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ MinMaxAvgStatistics that = (MinMaxAvgStatistics) o;
+ return minimum == that.minimum &&
+ maximum == that.maximum &&
+ average == that.average;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(minimum, maximum, average);
+ }
+}