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..83290555 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 @@ -31,8 +31,9 @@ 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.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 +48,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; @@ -83,6 +85,9 @@ public class JobVertexWatermarksHandler return new MetricCollectionResponseBody(Collections.emptyList()); } + final boolean includeMissingMetrics = + request.getQueryParameter(IncludeMissingQueryParameter.class) + .contains(Boolean.TRUE); AccessExecutionVertex[] taskVertices = jobVertex.getTaskVertices(); List metrics = new ArrayList<>(taskVertices.length); @@ -92,7 +97,7 @@ public class JobVertexWatermarksHandler + "." + MetricNames.IO_CURRENT_INPUT_WATERMARK; String watermarkValue = taskMetricStore.getMetric(id); - if (watermarkValue != null) { + if (watermarkValue != null || includeMissingMetrics) { 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..3d059e51 --- /dev/null +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/rest/messages/job/metrics/IncludeMissingQueryParameter.java @@ -0,0 +1,47 @@ +/* + * 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.MessageParameter; +import org.apache.flink.runtime.rest.messages.MessageQueryParameter; + +/** Query parameter for including missing watermark metrics in the response. */ +public class IncludeMissingQueryParameter extends MessageQueryParameter { + + private static final String QUERY_PARAMETER_NAME = "includeMissing"; + + public IncludeMissingQueryParameter() { + super(QUERY_PARAMETER_NAME, MessageParameter.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."; + } +} 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..532fdb4e 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(); @@ -72,7 +73,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..adecc508 --- /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. 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.JobVertexMessageParameters; +import org.apache.flink.runtime.rest.messages.MessageQueryParameter; + +import java.util.Collection; +import java.util.Collections; + +/** Message parameters for job vertex watermark REST handlers. */ +public class JobVertexWatermarksMessageParameters extends JobVertexMessageParameters { + + public final IncludeMissingQueryParameter includeMissingParameter = + new IncludeMissingQueryParameter(); + + @Override + public Collection> getQueryParameters() { + return Collections.singleton(includeMissingParameter); + } +} 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..ff416c5e 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; @@ -45,6 +45,7 @@ import javax.annotation.Nullable; import java.time.Duration; import java.util.Collections; +import java.util.List; import java.util.HashMap; import java.util.Map; import java.util.Objects; @@ -66,6 +67,7 @@ class JobVertexWatermarksHandlerTest { private MetricStore.TaskMetricStore taskMetricStore; private JobVertexWatermarksHandler watermarkHandler; private HandlerRequest request; + private Map pathParameters; private AccessExecutionJobVertex vertex; @BeforeEach @@ -90,17 +92,11 @@ class JobVertexWatermarksHandlerTest { NoOpExecutionGraphCache.INSTANCE, Mockito.mock(Executor.class)); - final Map pathParameters = new HashMap<>(); + pathParameters = new HashMap<>(); pathParameters.put(JobIDPathParameter.KEY, TEST_JOB_ID.toString()); pathParameters.put(JobVertexIdPathParameter.KEY, TEST_VERTEX_ID.toString()); - request = - HandlerRequest.resolveParametersAndCreate( - EmptyRequestBody.getInstance(), - new JobVertexMessageParameters(), - pathParameters, - Collections.emptyMap(), - Collections.emptyList()); + request = createRequest(Collections.emptyMap()); vertex = Mockito.mock(AccessExecutionJobVertex.class); Mockito.when(vertex.getJobVertexId()).thenReturn(TEST_VERTEX_ID); @@ -155,6 +151,33 @@ class JobVertexWatermarksHandlerTest { assertThat(response.getMetrics()).isEmpty(); } + @Test + void testMissingWatermarksIncludedWhenRequested() throws Exception { + request = createRequest(Collections.singletonMap("includeMissing", Collections.singletonList("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( + contains( + new MetricMatcher("0.currentInputWatermark", "23"), + new MetricMatcher("1.currentInputWatermark", null)))); + } + + private HandlerRequest createRequest( + Map> queryParameters) throws Exception { + return HandlerRequest.resolveParametersAndCreate( + EmptyRequestBody.getInstance(), + new JobVertexWatermarksMessageParameters(), + pathParameters, + queryParameters, + Collections.emptyList()); + } + private static class MetricMatcher extends BaseMatcher { private String id;