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<MetricCollectionResponseBody, JobVertexMessageParameters> {
+        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<Metric> 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<Boolean> {
+
+    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<MessageQueryParameter<?>> 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<EmptyRequestBody> request;
+    private Map<String, String> pathParameters;
     private AccessExecutionJobVertex vertex;
 
     @BeforeEach
@@ -90,17 +92,11 @@ class JobVertexWatermarksHandlerTest {
                         NoOpExecutionGraphCache.INSTANCE,
                         Mockito.mock(Executor.class));
 
-        final Map<String, String> 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<EmptyRequestBody> createRequest(
+            Map<String, List<String>> queryParameters) throws Exception {
+        return HandlerRequest.resolveParametersAndCreate(
+                EmptyRequestBody.getInstance(),
+                new JobVertexWatermarksMessageParameters(),
+                pathParameters,
+                queryParameters,
+                Collections.emptyList());
+    }
+
     private static class MetricMatcher extends BaseMatcher<Metric> {
 
         private String id;
