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<MetricCollectionResponseBody, JobVertexMessageParameters> {
+        extends AbstractJobVertexHandler<
+                MetricCollectionResponseBody, JobVertexWatermarksMessageParameters> {
 
     private final MetricFetcher metricFetcher;
 
@@ -85,14 +88,17 @@ public class JobVertexWatermarksHandler
 
         AccessExecutionVertex[] taskVertices = jobVertex.getTaskVertices();
         List<Metric> 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<Boolean> {
+
+    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<MessageQueryParameter<?>> 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<EmptyRequestBody> createRequestWithQueryParameter(
+            String key, String value) throws Exception {
+        final Map<String, java.util.List<String>> queryParameters = new HashMap<>();
+        queryParameters.put(key, Collections.singletonList(value));
+        return HandlerRequest.resolveParametersAndCreate(
+                EmptyRequestBody.getInstance(),
+                new JobVertexWatermarksMessageParameters(),
+                buildPathParameters(),
+                queryParameters,
+                Collections.emptyList());
+    }
+
+    private Map<String, String> buildPathParameters() {
+        final Map<String, String> 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<Metric> {
 
         private String id;
