source
.map(new MapFunction<String, Tuple2<String, Integer>>() {
    @Override
    public Tuple2<String, Integer> map(String value) {
        return Tuple2.of(value, 1);
    }
})
.keyBy(value -> value.f0)
.window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
.process(new WindowFunctionExample())
;

   private <R> WindowOperator<K, T, ?, R, W> apply(
            InternalWindowFunction<Iterable<T>, R, K, W> function) {
        if (evictor != null) {
            return buildEvictingWindowOperator(function);
        } else {
            ListStateDescriptor<T> stateDesc =
                    new ListStateDescriptor<>(
                            WINDOW_STATE_NAME, inputType.createSerializer(config));

            return buildWindowOperator(stateDesc, function);
        }
    }

public abstract class StreamElement {
  //判断是否是Watermark
  public final boolean isWatermark() {
    return getClass() == Watermark.class;
  }
  //判断是否为StreamStatus
  public final boolean isStreamStatus() {
    return getClass() == StreamStatus.class;
  }
  //判断是否为StreamRecord
  public final boolean isRecord() {
    return getClass() == StreamRecord.class;
  }
  //判断是否为LatencyMarker
  public final boolean isLatencyMarker() {
    return getClass() == LatencyMarker.class;
  }
  //转换为StreamRecord
  public final <E> StreamRecord<E> asRecord() {
    return (StreamRecord<E>) this;
  }
  //转换为Watermark
  public final Watermark asWatermark() {
    return (Watermark) this;
  }
  //转换为StreamStatus
  public final StreamStatus asStreamStatus() {
    return (StreamStatus) this;
  }
  //转换为LatencyMarker
  public final LatencyMarker asLatencyMarker() {
    return (LatencyMarker) this;
  }
}

@PublicEvolving
public final class Watermark extends StreamElement {
  /*The watermark that signifies end-of-event-time. */
  public static final Watermark MAX_WATERMARK = new Watermark(Long.MAX_VALUE);
  /* The timestamp of the watermark in milliseconds. */
  private final long timestamp;
  /* Creates a new watermark with the given timestamp in milliseconds.*/
  public Watermarklong timestamp) {
	this.timestamp = timestamp;
  }
  /*Returns the timestamp associated with this {@link Watermark} in milliseconds.**/
  public long getTimestamp() {
    return timestamp;
  }
}

SingleOutputStreamOperator<String> resultStream = socketStream
                .map((MapFunction<String, Product>) value -> {
                    String[] parts = value.split(",");
                    String name = parts[0];
                    long timestamp = Long.parseLong(parts[1]);
                    return new Product(name, timestamp);
                })
                // 指定 Watermark 生成策略（最大允许 10 天的乱序）
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<Product>forBoundedOutOfOrderness(Duration.ofDays(10))
                                .withTimestampAssigner((event, timestamp) -> event.productionTimestamp)
                )
                .keyBy(product -> product.productName)  // 按商品名分组
                .window(TumblingEventTimeWindows.of(Time.days(1))) // 窗口大小 1 天
                .sideOutputLateData(expiredTag)  // 过期商品进入侧输出流
                .process(new ExpiryCheckProcessWindowFunction());  // 窗口计算逻辑

.assignTimestampsAndWatermarks(
  WatermarkStrategy
  .<Product>forBoundedOutOfOrderness(Duration.ofDays(10))
  .withTimestampAssigner((event, timestamp) -> event.productionTimestamp)
)

@Public
public interface WatermarkStrategy<T>
        extends TimestampAssignerSupplier<T>, WatermarkGeneratorSupplier<T> {

    // ------------------------------------------------------------------------
    //  Methods that implementors need to implement.
    // ------------------------------------------------------------------------

    /** Instantiates a WatermarkGenerator that generates watermarks according to this strategy. */
    @Override
    WatermarkGenerator<T> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context);

    /**
     * Instantiates a {@link TimestampAssigner} for assigning timestamps according to this strategy.
     */
    @Override
    default TimestampAssigner<T> createTimestampAssigner(
            TimestampAssignerSupplier.Context context) {
        // By default, this is {@link RecordTimestampAssigner},
        // for cases where records come out of a source with valid timestamps, for example from
        // Kafka.
        return new RecordTimestampAssigner<>();
    }

    /**
     * Provides configuration for watermark alignment of a maximum watermark of multiple
     * sources/tasks/partitions in the same watermark group. The group may contain completely
     * independent sources (e.g. File and Kafka).
     *
     * <p>Once configured Flink will "pause" consuming from a source/task/partition that is ahead of
     * the emitted watermark in the group by more than the maxAllowedWatermarkDrift.
     */
    @PublicEvolving
    default WatermarkAlignmentParams getAlignmentParameters() {
        return WatermarkAlignmentParams.WATERMARK_ALIGNMENT_DISABLED;
    }

    // ------------------------------------------------------------------------
    //  Builder methods for enriching a base WatermarkStrategy
    // ------------------------------------------------------------------------

    /**
     * Creates a new {@code WatermarkStrategy} that wraps this strategy but instead uses the given
     * {@link TimestampAssigner} (via a {@link TimestampAssignerSupplier}).
     *
     * <p>You can use this when a {@link TimestampAssigner} needs additional context, for example
     * access to the metrics system.
     *
     * <pre>
     * {@code WatermarkStrategy<Object> wmStrategy = WatermarkStrategy
     *   .forMonotonousTimestamps()
     *   .withTimestampAssigner((ctx) -> new MetricsReportingAssigner(ctx));
     * }</pre>
     */
    default WatermarkStrategy<T> withTimestampAssigner(
            TimestampAssignerSupplier<T> timestampAssigner) {
        checkNotNull(timestampAssigner, "timestampAssigner");
        return new WatermarkStrategyWithTimestampAssigner<>(this, timestampAssigner);
    }

    /**
     * Creates a new {@code WatermarkStrategy} that wraps this strategy but instead uses the given
     * {@link SerializableTimestampAssigner}.
     *
     * <p>You can use this in case you want to specify a {@link TimestampAssigner} via a lambda
     * function.
     *
     * <pre>
     * {@code WatermarkStrategy<CustomObject> wmStrategy = WatermarkStrategy
     *   .<CustomObject>forMonotonousTimestamps()
     *   .withTimestampAssigner((event, timestamp) -> event.getTimestamp());
     * }</pre>
     */
    default WatermarkStrategy<T> withTimestampAssigner(
            SerializableTimestampAssigner<T> timestampAssigner) {
        checkNotNull(timestampAssigner, "timestampAssigner");
        return new WatermarkStrategyWithTimestampAssigner<>(
                this, TimestampAssignerSupplier.of(timestampAssigner));
    }

    /**
     * Creates a new enriched {@link WatermarkStrategy} that also does idleness detection in the
     * created {@link WatermarkGenerator}.
     *
     * <p>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.
     */
    default WatermarkStrategy<T> withIdleness(Duration idleTimeout) {
        checkNotNull(idleTimeout, "idleTimeout");
        checkArgument(
                !(idleTimeout.isZero() || idleTimeout.isNegative()),
                "idleTimeout must be greater than zero");
        return new WatermarkStrategyWithIdleness<>(this, idleTimeout);
    }

    /**
     * Creates a new {@link WatermarkStrategy} that configures the maximum watermark drift from
     * other sources/tasks/partitions in the same watermark group. The group may contain completely
     * independent sources (e.g. File and Kafka).
     *
     * <p>Once configured Flink will "pause" consuming from a source/task/partition that is ahead of
     * the emitted watermark in the group by more than the maxAllowedWatermarkDrift.
     *
     * @param watermarkGroup A group of sources to align watermarks
     * @param maxAllowedWatermarkDrift Maximal drift, before we pause consuming from the
     *     source/task/partition
     */
    @PublicEvolving
    default WatermarkStrategy<T> withWatermarkAlignment(
            String watermarkGroup, Duration maxAllowedWatermarkDrift) {
        return withWatermarkAlignment(
                watermarkGroup,
                maxAllowedWatermarkDrift,
                WatermarksWithWatermarkAlignment.DEFAULT_UPDATE_INTERVAL);
    }

    /**
     * Creates a new {@link WatermarkStrategy} that configures the maximum watermark drift from
     * other sources/tasks/partitions in the same watermark group. The group may contain completely
     * independent sources (e.g. File and Kafka).
     *
     * <p>Once configured Flink will "pause" consuming from a source/task/partition that is ahead of
     * the emitted watermark in the group by more than the maxAllowedWatermarkDrift.
     *
     * @param watermarkGroup A group of sources to align watermarks
     * @param maxAllowedWatermarkDrift Maximal drift, before we pause consuming from the
     *     source/task/partition
     * @param updateInterval How often tasks should notify coordinator about the current watermark
     *     and how often the coordinator should announce the maximal aligned watermark.
     */
    @PublicEvolving
    default WatermarkStrategy<T> withWatermarkAlignment(
            String watermarkGroup, Duration maxAllowedWatermarkDrift, Duration updateInterval) {
        return new WatermarksWithWatermarkAlignment<T>(
                this, watermarkGroup, maxAllowedWatermarkDrift, updateInterval);
    }

    // ------------------------------------------------------------------------
    //  Convenience methods for common watermark strategies
    // ------------------------------------------------------------------------

    /**
     * Creates a watermark strategy for situations with monotonously ascending timestamps.
     *
     * <p>The watermarks are generated periodically and tightly follow the latest timestamp in the
     * data. The delay introduced by this strategy is mainly the periodic interval in which the
     * watermarks are generated.
     *
     * @see AscendingTimestampsWatermarks
     */
    static <T> WatermarkStrategy<T> forMonotonousTimestamps() {
        return (ctx) -> new AscendingTimestampsWatermarks<>();
    }

    /**
     * Creates a watermark strategy for situations where records are out of order, but you can place
     * an upper bound on how far the events are out of order. An out-of-order bound B means that
     * once the an event with timestamp T was encountered, no events older than {@code T - B} will
     * follow any more.
     *
     * <p>The watermarks are generated periodically. The delay introduced by this watermark strategy
     * is the periodic interval length, plus the out of orderness bound.
     *
     * @see BoundedOutOfOrdernessWatermarks
     */
    static <T> WatermarkStrategy<T> forBoundedOutOfOrderness(Duration maxOutOfOrderness) {
        return (ctx) -> new BoundedOutOfOrdernessWatermarks<>(maxOutOfOrderness);
    }

    /** Creates a watermark strategy based on an existing {@link WatermarkGeneratorSupplier}. */
    static <T> WatermarkStrategy<T> forGenerator(WatermarkGeneratorSupplier<T> generatorSupplier) {
        return generatorSupplier::createWatermarkGenerator;
    }

    /**
     * Creates a watermark strategy that generates no watermarks at all. This may be useful in
     * scenarios that do pure processing-time based stream processing.
     */
    static <T> WatermarkStrategy<T> noWatermarks() {
        return (ctx) -> new NoWatermarksGenerator<>();
    }
}

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 设置 Watermark 生成周期为 2s
env.getConfig().setAutoWatermarkInterval(2000);

// 输入数据格式：时间戳，单词，频次
// 例如：1000,a,1
DataStreamSource<String> source = env.socketTextStream("192.168.117.128", 9999);

// 定义 Watermark 策略
WatermarkStrategy<String> strategy =
        WatermarkStrategy.<String>forGenerator(new MyWatermarkGeneratorSupplier<>(Duration.ofSeconds(3)))
                .withTimestampAssigner(context -> (s, l) -> Long.parseLong(s.split(",")[0]));

// 业务处理：transformation
source
        .assignTimestampsAndWatermarks(strategy)
        .map(value -> {
            String[] splits = value.split(",");
            return Tuple2.of(splits[1].trim(), Integer.parseInt(splits[2].trim()));
        })
        .returns(Types.TUPLE(Types.STRING, Types.INT))
        .keyBy(value -> value.f0)
        .window(TumblingEventTimeWindows.of(Time.seconds(5)))
        .reduce((Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) -> {
            System.out.println("-----reduce invoked----" + value1.f0 + "==>" + (value1.f1 + value2.f1));
            return Tuple2.of(value1.f0, value1.f1 + value2.f1);
        }, new ProcessWindowFunction<Tuple2<String, Integer>, String, String, TimeWindow>() {

            @Override
            public void process(String s, Context context, Iterable<Tuple2<String, Integer>> iterable, Collector<String> collector) {

                FastDateFormat format = FastDateFormat.getInstance("yyyy-MM-dd HH:mm:ss.SSS");

                for (Tuple2<String, Integer> element : iterable) {
                    collector.collect("[" + format.format(context.window().getStart()) + "==>" +
                            format.format(context.window().getEnd()) + "], " + element.f0 + "==>" + element.f1);
                }
            }
        })
        .print();

env.execute("Watermark App");

public class MyWatermarkGeneratorSupplier<T> implements WatermarkGeneratorSupplier<T> {

    private final Duration duration;

    public MyWatermarkGeneratorSupplier(Duration duration) {
        this.duration = duration;
    }

    @Override
    public WatermarkGenerator<T> createWatermarkGenerator(Context context) {
        return new MyWatermarkGenerator<>(duration);
    }
}

public class MyWatermarkGenerator<T> implements WatermarkGenerator<T> {

    private static final Logger LOGGER = LoggerFactory.getLogger(MyWatermarkGenerator.class);

    private long maxTimestamp;
    private final long outOfOrdernessMillis;

    public MyWatermarkGenerator(Duration maxOutOfOrderness) {
        Preconditions.checkNotNull(maxOutOfOrderness, "maxOutOfOrderness");
        Preconditions.checkArgument(!maxOutOfOrderness.isNegative(), "maxOutOfOrderness cannot be negative");
        this.outOfOrdernessMillis = maxOutOfOrderness.toMillis();
        this.maxTimestamp = -9223372036854775808L + this.outOfOrdernessMillis + 1L;
    }

    public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
        this.maxTimestamp = Math.max(this.maxTimestamp, eventTimestamp);
    }

    public void onPeriodicEmit(WatermarkOutput output) {
        Watermark watermark = new Watermark(this.maxTimestamp - this.outOfOrdernessMillis - 1L);
        output.emitWatermark(watermark);
        LOGGER.info("current watermark ==> {}", watermark.getTimestamp());
    }
}

今天,1741968000000

p1,1741795200000
p2,1741622400000
p3,1740758400000
p4,1741017600000
p5,1741104000000

第二天,1742054400000

3月21日,1742486400000

3月22日,1742572800000

数据1（时间戳：12:00:00）到达，水位线设置为12:00:00 - 2秒 = 11:59:58。

数据2（时间戳：12:00:05）到达，水位线更新为12:00:05 - 2秒 = 12:00:03。

数据3（时间戳：12:00:12）到达，水位线更新为12:00:12 - 2秒 = 12:00:10。

public class MyWatermarkGenerator<T> implements WatermarkGenerator<T> {

    private static final Logger LOGGER = LoggerFactory.getLogger(MyWatermarkGenerator.class);

    private long maxTimestamp;
    private final long outOfOrdernessMillis;

    public MyWatermarkGenerator(Duration maxOutOfOrderness) {
        Preconditions.checkNotNull(maxOutOfOrderness, "maxOutOfOrderness");
        Preconditions.checkArgument(!maxOutOfOrderness.isNegative(), "maxOutOfOrderness cannot be negative");
        this.outOfOrdernessMillis = maxOutOfOrderness.toMillis();
        this.maxTimestamp = -9223372036854775808L + this.outOfOrdernessMillis + 1L;
    }

    public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
        this.maxTimestamp = Math.max(this.maxTimestamp, eventTimestamp);
    }

    public void onPeriodicEmit(WatermarkOutput output) {
        Watermark watermark = new Watermark(this.maxTimestamp - this.outOfOrdernessMillis - 1L);
        output.emitWatermark(watermark);
        LOGGER.info("current watermark ==> {}", watermark.getTimestamp());
    }
}

@Public
public class BoundedOutOfOrdernessWatermarks<T> implements WatermarkGenerator<T> {

    /** The maximum timestamp encountered so far. */
    private long maxTimestamp;

    /** The maximum out-of-orderness that this watermark generator assumes. */
    private final long outOfOrdernessMillis;

    /**
     * Creates a new watermark generator with the given out-of-orderness bound.
     *
     * @param maxOutOfOrderness The bound for the out-of-orderness of the event timestamps.
     */
    public BoundedOutOfOrdernessWatermarks(Duration maxOutOfOrderness) {
        checkNotNull(maxOutOfOrderness, "maxOutOfOrderness");
        checkArgument(!maxOutOfOrderness.isNegative(), "maxOutOfOrderness cannot be negative");

        this.outOfOrdernessMillis = maxOutOfOrderness.toMillis();

        // start so that our lowest watermark would be Long.MIN_VALUE.
        this.maxTimestamp = Long.MIN_VALUE + outOfOrdernessMillis + 1;
    }

    // ------------------------------------------------------------------------

    @Override
    public void onEvent(T event, long eventTimestamp, WatermarkOutput output) {
        maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
    }

    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        output.emitWatermark(new Watermark(maxTimestamp - outOfOrdernessMillis - 1));
    }
}

env.getConfig().setAutoWatermarkInterval(400L);

static <T> WatermarkStrategy<T> forMonotonousTimestamps() {
    return (ctx) -> new AscendingTimestampsWatermarks<>();
}

@Public
public class AscendingTimestampsWatermarks<T> extends BoundedOutOfOrdernessWatermarks<T> {

    /** Creates a new watermark generator with for ascending timestamps. */
    public AscendingTimestampsWatermarks() {
        super(Duration.ofMillis(0));
    }
}

OutputTag<Product> expiredTag = new OutputTag<Product>("expired") {};
socketStream
  .map((MapFunction<String, Product>) value -> {
      String[] parts = value.split(",");
      String name = parts[0];
      long timestamp = Long.parseLong(parts[1]);
      return new Product(name, timestamp);
  })
  // 指定 Watermark 生成策略（最大允许 10 天的乱序）
  .assignTimestampsAndWatermarks(
          WatermarkStrategy
                  .<Product>forBoundedOutOfOrderness(Duration.ofDays(10))
                  .withTimestampAssigner((event, timestamp) -> event.productionTimestamp)
  )
  .keyBy(product -> product.productName)  // 按商品名分组
  .window(TumblingEventTimeWindows.of(Time.days(1))) // 窗口大小 1 天
  .allowedLateness(Time.seconds(2)) // 推迟2s关窗
  .sideOutputLateData(expiredTag)  // 过期商品进入侧输出流
  .process(new ExpiryCheckProcessWindowFunction());  // 窗口计算逻辑