[FLINK-17899][core][refactor] Add a utility NoWatermarksGenerator

flink-core/src/main/java/org/apache/flink/api/common/eventtime/NoWatermarksGenerator.java

@@ -0,0 +1,34 @@
+/*
+ * 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.api.common.eventtime;
+
+import org.apache.flink.annotation.Public;
+
+/**
+ * An implementation of a {@link WatermarkGenerator} that generates no Watermarks.
+ */
+@Public
+public final class NoWatermarksGenerator<E> implements WatermarkGenerator<E> {
+
+	@Override
+	public void onEvent(E event, long eventTimestamp, WatermarkOutput output) {}
+
+	@Override
+	public void onPeriodicEmit(WatermarkOutput output) {}
+}

flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkStrategies.java

@@ -175,6 +175,14 @@ public static <T> WatermarkStrategies<T> forGenerator(WatermarkGeneratorSupplier
 		return new WatermarkStrategies<>(new FromWatermarkGeneratorSupplier<>(generatorSupplier));
 	}
 
+	/**
+	 * Starts building a watermark strategy that generates no watermarks at all.
+	 * This may be useful in scenarios that do pure processing-time based stream processing.
+	 */
+	public static <T> WatermarkStrategies<T> noWatermarks() {
+		return new WatermarkStrategies<>((ctx) -> new NoWatermarksGenerator<>());
+	}
+
 	// ------------------------------------------------------------------------
 
 	private static final class FromWatermarkGeneratorSupplier<T> implements WatermarkStrategy<T> {