Flink时态连接执行结果不一致问题排查求助

本地单节点Kraft模式Kafka（v7.7.1）部署了三个单分区主题：optionsTopic、stocksTopic和referencesTopic。基于Flink（v1.20）开发应用，通过时态连接合并消息并写入processedOptionsTopic，但重复执行应用（重放相同数据集、全新环境）时结果不一致：有时所有记录匹配正常，有时stockTrades中已存在的symb关联结果为null。

数据源定义 #

optionTrades（Kafka连接器）

// new options produced every 1s 
tEnv.createTemporaryTable("optionTrades", TableDescriptor.forConnector("kafka")
    .schema(Schema.newBuilder()
        .fromColumns(SchemaBuilders.forTrades())
        .watermark("tTime", "tTime - INTERVAL '5' SECOND")
        .build())
    .option("topic", "optionsTopic")
    .option("properties.bootstrap.servers", kafkaBServers)
    .option("format", "json")
    .option("scan.startup.mode", "earliest-offset")
    .option("json.timestamp-format.standard", "ISO-8601")
    .build());

stockTrades（Upsert-Kafka连接器）

// new stocks produced every 1s 
tEnv.createTemporaryTable("stockTrades", TableDescriptor.forConnector("upsert-kafka")
    .schema(Schema.newBuilder()
        .fromColumns(SchemaBuilders.forTrades())
        .watermark("tTime", "tTime - INTERVAL '5' SECOND")
        .primaryKey("symb")
        .build())
    .option("topic", "stocksTopic")
    .option("properties.bootstrap.servers", kafkaBServers)
    .option("key.format", "raw")
    .option("value.format", "json")
    .option("value.json.timestamp-format.standard", "ISO-8601")
    .build());

tickersReference（Upsert-Kafka连接器）

// new refs are produced every 30 min
tEnv.createTemporaryTable("tickersReference", TableDescriptor.forConnector("upsert-kafka")
    .schema(Schema.newBuilder()
        .fromColumns(SchemaBuilders.forRefs())
        .watermark("rtTime", "rtTime - INTERVAL '3' SECOND")
        .primaryKey("optionSymb")
        .build())
    .option("topic", "referencesTopic")
    .option("properties.bootstrap.servers", kafkaBServers)
    .option("key.format", "raw")
    .option("value.format", "json")
    .option("value.json.timestamp-format.standard", "ISO-8601")
    .build());

输出端定义 #

tEnv.createTemporaryTable("processedOptions", TableDescriptor.forConnector("kafka")
    .schema(Schema.newBuilder()
        .fromColumns(SchemaBuilders.forProcessedOptionTrades())
        .build())
    .option("topic", "processedOptionsTopic")
    .option("properties.bootstrap.servers", kafkaBServers)
    .option("sink.delivery-guarantee", "exactly-once")
    .option("sink.transactional-id-prefix", "exactlyOncePrefix")
    .option("format", "json")
    .option("json.timestamp-format.standard", "ISO-8601")
    .option("properties.transaction.timeout.ms", "900000")
    .build());

时态连接逻辑 #

Table mergedStreams = tEnv.sqlQuery(
    "SELECT optionTrades.*, " +
        // other fields were omitted for conciseness... 
        "tickersReference.rtTime AS rtTime " +
    "FROM optionTrades " +
    "LEFT JOIN tickersReference " +
        "FOR SYSTEM_TIME AS OF optionTrades.tTime " +
        "ON optionTrades.symb = tickersReference.optionSymb"
);

tEnv.createTemporaryView("optionTradeAndReference", mergedStreams);
tEnv.executeSql("INSERT INTO processedOptions " +
    "SELECT optionTradeAndReference.*, " +
        "stockTrades.tId AS stId, " +
        // other fields were omitted for conciseness... 
        "stockTrades.tTime AS stTime " +
    "FROM optionTradeAndReference " +
    "LEFT JOIN stockTrades " +
        "FOR SYSTEM_TIME AS OF optionTradeAndReference.tTime " +
        "ON optionTradeAndReference.stockSymb = stockTrades.symb");

已设置table.exec.source.idle-timeout = 500 ms避免referencesTopic空闲，调整过watermark，parallelism.default设为1，问题仍存在。

1. 检查时态表的事件时间对齐与数据到达顺序 #

时态连接依赖左表事件时间点时态表的快照状态，如果stockTrades或tickersReference中对应key的更新记录晚于optionTrades的事件时间到达Flink，就会出现关联结果为null的情况：

• 开启Flink的事件时间调试日志（将org.apache.flink.table.runtime.operators.join.temporal日志级别设为DEBUG），查看每次时态连接时，对应key在目标时间点是否已有数据。
• 验证Kafka主题中stockTrades的消息实际写入时间是否确实早于对应的optionTrades事件时间，本地IO波动可能导致业务逻辑上的顺序与实际消息顺序不一致。

2. 移除时态表的watermark并调整状态TTL #

时态表（维度表）不需要watermark，多余的watermark可能干扰状态保留逻辑：

• 删除stockTrades和tickersReference上的watermark定义，仅保留optionTrades的watermark用于推进流的事件时间。
• 显式设置全局状态TTL，例如table.exec.state.ttl = 86400000（24小时），确保重放数据时状态不会被提前清理。

3. 验证第一次时态连接的映射结果 #

第二次连接stockTrades的关联键stockSymb来自第一次与tickersReference的连接，需确认该字段是否有效：

• 在第一步查询中添加日志输出，查看stockSymb是否为null，区分是映射失败还是stockTrades关联失败。
• 检查tickersReference中数据的rtTime是否早于对应的optionTrades.tTime，如果optionTrades事件时间早于映射数据的生成时间，会导致stockSymb为null。

4. 确保Kafka消费的严格顺序 #

即使是单分区主题，Kafka消费者参数可能导致消费批次顺序波动：

• 在Flink的Kafka连接器配置中添加option("properties.fetch.wait.max.ms", "0")，让消费者立即返回可用数据，减少批次等待带来的顺序不确定性。
• 配置option("properties.isolation.level", "read_committed")，确保消费已提交的消息，避免未提交事务消息干扰。

5. 检查Exactly-Once语义的事务与状态恢复 #

输出端启用exactly-once后，旧事务残留或状态恢复异常可能导致结果不一致：

• 重放前清理processedOptionsTopic的事务日志，避免旧事务影响新写入。
• 将状态后端切换为RocksDB并启用增量Checkpoint，内存状态在重启时可能存在数据丢失或顺序错乱的风险。

内容的提问来源于stack exchange，提问作者Joseandro Luiz