dawidwys commented on a change in pull request #12147: URL: https://github.com/apache/flink/pull/12147#discussion_r425131321
########## File path: flink-core/src/main/java/org/apache/flink/api/common/eventtime/WatermarkStrategies.java ########## @@ -0,0 +1,230 @@ +/* + * 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.PublicEvolving; +import org.apache.flink.util.FlinkRuntimeException; +import org.apache.flink.util.InstantiationUtil; + +import javax.annotation.Nullable; + +import java.io.Serializable; +import java.time.Duration; + +import static org.apache.flink.util.Preconditions.checkArgument; +import static org.apache.flink.util.Preconditions.checkNotNull; + +/** + * WatermarkStrategies is a simply way to build a {@link WatermarkStrategy} by configuring + * common strategies. + */ +@PublicEvolving +public final class WatermarkStrategies<T> { + + /** + * The {@link TimestampAssigner} to use. This can be {@code null} for cases where records come + * out of a source with valid timestamps, for example from Kafka. + */ + @Nullable + private TimestampAssigner<T> timestampAssigner = null; + + /** The base strategy for watermark generation. Starting point, is always set. */ + private final WatermarkStrategy<T> baseStrategy; + + /** Optional idle timeout for watermarks. */ + @Nullable + private Duration idleTimeout; + + private WatermarkStrategies(WatermarkStrategy<T> baseStrategy) { + this.baseStrategy = baseStrategy; + } + + // ------------------------------------------------------------------------ + // builder methods + // ------------------------------------------------------------------------ + + /** + * Add an idle timeout to the watermark strategy. + * If no records flow in a partition of a stream for that amount of time, then that partition + * is considered "idle" and will not hold back the progress of watermarks in downstream operators. + * + * <p>Idleness can be important if some partitions have little data and might not have events during + * some periods. Without idleness, these streams can stall the overall event time progress of the + * application. + */ + public WatermarkStrategies<T> withIdleness(Duration idleTimeout) { + checkNotNull(idleTimeout, "idleTimeout"); + checkArgument(!(idleTimeout.isZero() || idleTimeout.isNegative()), "idleTimeout must be greater than zero"); + this.idleTimeout = idleTimeout; + return this; + } + + /** + * Adds the given {@link TimestampAssigner} to this {@link WatermarkStrategies}. + */ + public <TA extends TimestampAssigner<T> & Serializable> WatermarkStrategies<T> withTimestampAssigner(TA timestampAssigner) { + checkNotNull(timestampAssigner, "timestampAssigner"); + this.timestampAssigner = timestampAssigner; + return this; + } + + /** + * Adds the given {@link TimestampAssigner} to this {@link WatermarkStrategies}. + */ + public WatermarkStrategies<T> withTimestampAssigner(SerializableTimestampAssigner<T> timestampAssigner) { + checkNotNull(timestampAssigner, "timestampAssigner"); + this.timestampAssigner = timestampAssigner; + return this; + } + + /** + * Build the watermark strategy. + */ + public WatermarkStrategy<T> build() { + WatermarkStrategy<T> strategy = this.baseStrategy; + + if (idleTimeout != null) { + strategy = new WithIdlenessStrategy<>(strategy, idleTimeout); + } + + if (timestampAssigner != null) { + strategy = new WithTimestampAssigner<>(strategy, timestampAssigner); + } + + return strategy; + } + + // ------------------------------------------------------------------------ + // builder entry points + // ------------------------------------------------------------------------ Review comment: nit: A matter of personal taste, but I prefer the entry methods to appear first in the class. That way the class reads more naturally from the top to the bottom. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org
