Flink 1.13.2中Kafka消费流Join本地集合流无输出问题

你的代码存在3个核心逻辑错误，直接导致窗口无法触发输出Join结果：

• 重复为Kafka流分配Watermark，造成时间语义混乱
  你在构建dataStream1时，已经在fromSource方法中传入了自定义waterMark()策略，后续又调用rebalance().assignTimestampsAndWatermarks(waterMark())二次绑定水位线生成逻辑，会导致流的事件时间计算、水位线推进逻辑出现冲突，无法按预期推进窗口进度。
• 有限流的Watermark推进时序和Kafka流完全错位，是无输出的核心原因
  dataStream2是通过fromElements创建的有限数据集，作业启动瞬间所有元素就会被全部下发，元素发送完成后Flink会自动为该流注入值为Long.MAX_VALUE的终止水位线，标记流数据全部发送完毕。
  而事件时间窗口触发Join的必要条件是：两条流的水位线同时超过当前窗口的结束时间，且窗口触发时两条流中落在窗口时间范围内的数据都还在状态中未被清理。你的代码里存在致命时序差：
  1. 作业启动时dataStream2的3个元素瞬间完成下发，记录的事件时间是作业启动时刻的T0，很快终止水位线到达，所有关联到T0附近窗口的dataStream2数据会在窗口触发后被清理
  2. Kafka流的消息是作业启动后陆续到达的，这些消息的事件时间取的是处理时的System.currentTimeMillis()，也就是T1远大于T0，等Kafka流的水位线推进到对应滑动窗口的结束时间时，dataStream2在对应窗口里的数据早就被清理，根本无法完成匹配。
• 水位线策略选型不符合跨流Join的场景要求
  你使用的AscendingTimestampsWatermarks是严格单调递增的水位线生成器，零乱序容忍，直接取处理时刻的系统时间作为事件时间，两条流的时间进度完全没有对齐机制，进一步放大了数据错过窗口的概率。

修正方案 #

1. 删掉Kafka流重复的水位线配置，同一条流只绑定一次Watermark策略
2. 如果你是要做Kafka无限流和本地固定维度数据集的关联，不要用窗口Join：直接在算子中加载本地数据集到内存做常规关联，或者将本地数据集作为广播流做广播关联，完全规避水位线对齐、窗口触发的问题，性能和稳定性都更好。
3. 如果一定要用滑动窗口Join，调整水位线策略，给两条流设置合理的乱序容忍，同时不要让有限流的水位线提前冲至最大值，保证窗口生命周期内两边的数据都能被状态留存。

修正后的核心代码参考：

public static void main(String[] args) {
    KafkaSource<JsonNode> kafkaSource = ...
    final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
    
    // 移除fromSource中冗余的watermark配置，仅保留一次水位线分配
    final DataStream<JsonNode> dataStream1 = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "Kafka")
            .rebalance()
            .assignTimestampsAndWatermarks(waterMark());
    
    final DataStream<String> dataStream2 = env.fromElements("John", "Zbe", "Abe")
            .assignTimestampsAndWatermarks(waterMark());
    
    dataStream1
            .join(dataStream2)
            .where(new KeySelector<JsonNode, String>() {
                @Override
                public String getKey(JsonNode value) throws Exception {
                    return value.get("name").asText();
                }
            })
            .equalTo(new KeySelector<String, String>() {
                @Override
                public String getKey(String value) throws Exception {
                    return value;
                }
            })
            .window(SlidingEventTimeWindows.of(Time.minutes(50), Time.minutes(10)))
            .apply(new JoinFunction<JsonNode, String, String>() {
                @Override
                public String join(JsonNode first, String second) throws Exception {
                    return first+" "+second;
                }
            }).print();
            
    env.execute();
}

修正后的Watermark实现：

private static <T>  WatermarkStrategy<T> waterMark() {
    // 替换严格递增的水位线，设置10s乱序容忍适配跨流时间差
    return WatermarkStrategy
            .<T>forBoundedOutOfOrderness(Duration.ofSeconds(10))
            .withTimestampAssigner((event, timestamp) -> System.currentTimeMillis());
}

注意：窗口Join本身只适合两条都是无限流、且需要按时间窗口做范围匹配的场景，流和本地静态维度集关联用窗口Join属于选型错误。

内容的提问来源于stack exchange，提问作者Minh Trần