diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/JobVertexWatermarksHandler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/JobVertexWatermarksHandler.java index d341e411..d80ed097 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/JobVertexWatermarksHandler.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/handler/job/metrics/JobVertexWatermarksHandler.java @@ -29,10 +29,12 @@ import org.apache.flink.runtime.rest.handler.job.AbstractJobVertexHandler; import org.apache.flink.runtime.rest.handler.legacy.ExecutionGraphCache; import org.apache.flink.runtime.rest.handler.legacy.metrics.MetricFetcher; import org.apache.flink.runtime.rest.handler.legacy.metrics.MetricStore; +import org.apache.flink.runtime.rest.handler.util.HandlerRequestUtils; import org.apache.flink.runtime.rest.messages.EmptyRequestBody; import org.apache.flink.runtime.rest.messages.JobIDPathParameter; -import org.apache.flink.runtime.rest.messages.JobVertexMessageParameters; +import org.apache.flink.runtime.rest.messages.job.metrics.IncludeMissingQueryParameter; import org.apache.flink.runtime.rest.messages.job.metrics.JobVertexWatermarksHeaders; +import org.apache.flink.runtime.rest.messages.job.metrics.JobVertexWatermarksMessageParameters; import org.apache.flink.runtime.rest.messages.job.metrics.Metric; import org.apache.flink.runtime.rest.messages.job.metrics.MetricCollectionResponseBody; import org.apache.flink.runtime.webmonitor.RestfulGateway; @@ -47,7 +49,8 @@ import java.util.concurrent.Executor; /** Handler that returns the watermarks given a {@link JobID} and {@link JobVertexID}. */ public class JobVertexWatermarksHandler - extends AbstractJobVertexHandler { + extends AbstractJobVertexHandler< + MetricCollectionResponseBody, JobVertexWatermarksMessageParameters> { private final MetricFetcher metricFetcher; @@ -85,14 +88,17 @@ public class JobVertexWatermarksHandler AccessExecutionVertex[] taskVertices = jobVertex.getTaskVertices(); List metrics = new ArrayList<>(taskVertices.length); + final boolean includeMissing = + HandlerRequestUtils.getQueryParameter( + request, IncludeMissingQueryParameter.class, false); for (AccessExecutionVertex taskVertex : taskVertices) { String id = taskVertex.getParallelSubtaskIndex() + "." - + MetricNames.IO_CURRENT_INPUT_WATERMARK; + + MetricNames.IO_CURRENT_INPUT_WATERMARK; String watermarkValue = taskMetricStore.getMetric(id); - if (watermarkValue != null) { + if (watermarkValue != null || includeMissing) { metrics.add(new Metric(id, watermarkValue)); } } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/IncludeMissingQueryParameter.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/IncludeMissingQueryParameter.java new file mode 100644 index 00000000..f4697786 --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/IncludeMissingQueryParameter.java @@ -0,0 +1,44 @@ +/* + * 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.job.metrics; + +import org.apache.flink.runtime.rest.messages.MessageQueryParameter; + +/** Query parameter for including missing watermark metrics in the response. */ +public class IncludeMissingQueryParameter extends MessageQueryParameter { + + IncludeMissingQueryParameter() { + super("includeMissing", MessageParameterRequisiteness.OPTIONAL); + } + + @Override + public String convertValueToString(Boolean value) { + return String.valueOf(value); + } + + @Override + public Boolean convertStringToValue(String value) { + return Boolean.valueOf(value); + } + + @Override + public String getDescription() { + return "Boolean value that specifies whether missing watermark metrics should be included with null values."; + } +} diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/JobVertexWatermarksHeaders.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/JobVertexWatermarksHeaders.java index fc564105..2c8f34fd 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/JobVertexWatermarksHeaders.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/JobVertexWatermarksHeaders.java @@ -22,7 +22,6 @@ import org.apache.flink.runtime.rest.HttpMethodWrapper; import org.apache.flink.runtime.rest.messages.EmptyRequestBody; import org.apache.flink.runtime.rest.messages.JobIDPathParameter; import org.apache.flink.runtime.rest.messages.JobVertexIdPathParameter; -import org.apache.flink.runtime.rest.messages.JobVertexMessageParameters; import org.apache.flink.runtime.rest.messages.MessageHeaders; import org.apache.flink.runtime.rest.messages.RuntimeMessageHeaders; @@ -31,7 +30,9 @@ import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseSt /** {@link MessageHeaders} for retrieving watermarks. */ public final class JobVertexWatermarksHeaders implements RuntimeMessageHeaders< - EmptyRequestBody, MetricCollectionResponseBody, JobVertexMessageParameters> { + EmptyRequestBody, + MetricCollectionResponseBody, + JobVertexWatermarksMessageParameters> { public static final JobVertexWatermarksHeaders INSTANCE = new JobVertexWatermarksHeaders(); @@ -53,7 +54,8 @@ public final class JobVertexWatermarksHeaders @Override public String getDescription() { - return "Returns the watermarks for all subtasks of a task."; + return "Returns the watermarks for all subtasks of a task. Set includeMissing=true to" + + " include null-valued entries for subtasks whose watermark metric is missing."; } @Override @@ -72,7 +74,7 @@ public final class JobVertexWatermarksHeaders } @Override - public JobVertexMessageParameters getUnresolvedMessageParameters() { - return new JobVertexMessageParameters(); + public JobVertexWatermarksMessageParameters getUnresolvedMessageParameters() { + return new JobVertexWatermarksMessageParameters(); } } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/JobVertexWatermarksMessageParameters.java b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/JobVertexWatermarksMessageParameters.java new file mode 100644 index 00000000..acde1833 --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/JobVertexWatermarksMessageParameters.java @@ -0,0 +1,37 @@ +/* + * 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. This file is licensed 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.job.metrics; + +import org.apache.flink.runtime.rest.messages.JobVertexMessageParameters; +import org.apache.flink.runtime.rest.messages.MessageQueryParameter; + +import java.util.Arrays; +import java.util.Collection; + +/** Message parameters for job vertex watermark requests. */ +public class JobVertexWatermarksMessageParameters extends JobVertexMessageParameters { + + private final IncludeMissingQueryParameter includeMissingQueryParameter = + new IncludeMissingQueryParameter(); + + @Override + public Collection> getQueryParameters() { + return Arrays.asList(includeMissingQueryParameter); + } +} diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/metrics/JobVertexWatermarksHandlerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/metrics/JobVertexWatermarksHandlerTest.java index faf9714f..02da8819 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/metrics/JobVertexWatermarksHandlerTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/handler/job/metrics/JobVertexWatermarksHandlerTest.java @@ -28,7 +28,7 @@ import org.apache.flink.runtime.rest.handler.legacy.metrics.MetricStore; import org.apache.flink.runtime.rest.messages.EmptyRequestBody; import org.apache.flink.runtime.rest.messages.JobIDPathParameter; import org.apache.flink.runtime.rest.messages.JobVertexIdPathParameter; -import org.apache.flink.runtime.rest.messages.JobVertexMessageParameters; +import org.apache.flink.runtime.rest.messages.job.metrics.JobVertexWatermarksMessageParameters; import org.apache.flink.runtime.rest.messages.job.metrics.Metric; import org.apache.flink.runtime.rest.messages.job.metrics.MetricCollectionResponseBody; import org.apache.flink.runtime.rest.util.NoOpExecutionGraphCache; @@ -97,7 +97,7 @@ class JobVertexWatermarksHandlerTest { request = HandlerRequest.resolveParametersAndCreate( EmptyRequestBody.getInstance(), - new JobVertexMessageParameters(), + new JobVertexWatermarksMessageParameters(), pathParameters, Collections.emptyMap(), Collections.emptyList()); @@ -155,6 +155,57 @@ class JobVertexWatermarksHandlerTest { assertThat(response.getMetrics()).isEmpty(); } + @Test + void testMissingWatermarksIncludedWhenRequested() throws Exception { + request = + createRequestWithQueryParameter("includeMissing", "true"); + + Mockito.when(taskMetricStore.getMetric("0.currentInputWatermark")).thenReturn("23"); + Mockito.when(taskMetricStore.getMetric("1.currentInputWatermark")).thenReturn(null); + + MetricCollectionResponseBody response = watermarkHandler.handleRequest(request, vertex); + + assertThat(response.getMetrics()) + .satisfies( + matching( + containsInAnyOrder( + new MetricMatcher("0.currentInputWatermark", "23"), + new MetricMatcher("1.currentInputWatermark", null)))); + } + + @Test + void testMissingWatermarksNotIncludedWhenExplicitlyDisabled() throws Exception { + request = + createRequestWithQueryParameter("includeMissing", "false"); + + Mockito.when(taskMetricStore.getMetric("0.currentInputWatermark")).thenReturn("23"); + Mockito.when(taskMetricStore.getMetric("1.currentInputWatermark")).thenReturn(null); + + MetricCollectionResponseBody response = watermarkHandler.handleRequest(request, vertex); + + assertThat(response.getMetrics()) + .satisfies(matching(contains(new MetricMatcher("0.currentInputWatermark", "23")))); + } + + private HandlerRequest createRequestWithQueryParameter( + String key, String value) throws Exception { + final Map> queryParameters = new HashMap<>(); + queryParameters.put(key, Collections.singletonList(value)); + return HandlerRequest.resolveParametersAndCreate( + EmptyRequestBody.getInstance(), + new JobVertexWatermarksMessageParameters(), + buildPathParameters(), + queryParameters, + Collections.emptyList()); + } + + private Map buildPathParameters() { + final Map pathParameters = new HashMap<>(); + pathParameters.put(JobIDPathParameter.KEY, TEST_JOB_ID.toString()); + pathParameters.put(JobVertexIdPathParameter.KEY, TEST_VERTEX_ID.toString()); + return pathParameters; + } + private static class MetricMatcher extends BaseMatcher { private String id;