Apache Flink的KinesisStreamsSink写入错误时如何保证记录顺序

要解决写入错误时的批次顺序混乱问题，你需要从语义保障、重试策略、批次原子性三个维度调整KinesisStreamsSink的配置，确保批次必须完全处理完成（成功或故障恢复）后才会处理下一批次。以下是具体方案：

核心配置调整 #

1. 启用精确一次语义（Exactly-Once） #

精确一次语义是保障顺序的基础，它结合Flink检查点和Kinesis的幂等写入能力，确保只有当整个批次的所有记录都成功写入Kinesis后，Flink才会提交状态并处理下一批次。

2. 自定义重试策略应对临时错误 #

针对吞吐量超限这类临时错误，需要配置足够的重试次数和间隔，避免失败记录被直接混入下一批。当重试耗尽后，让任务故障转移（而不是继续处理新批次），这样Flink会从最近的检查点恢复，重新处理整个失败批次，不会和新批次混合。

3. 保留单请求并发限制 #

你已经设置的setMaxInFlightRequests(1)要保留，确保同一时间只有一个写入请求在执行，彻底避免批次并行发送导致的顺序混乱。

修改后的完整代码示例 #

KinesisStreamsSink.<Rec>builder()
        .setKinesisClientProperties(kinesisProducerConfig)
        .setSerializationSchema((SerializationSchema<Rec>) rec -> 
                rec.Json.getBytes(StandardCharsets.UTF_8))
        .setStreamName(streamName)
        .setPartitionKeyGenerator(element -> element.key)
        .setMaxInFlightRequests(1)
        // 启用精确一次语义，依赖Flink检查点开启
        .setExactlyOnce(true)
        // 配置重试策略：针对临时错误重试5次，每次间隔2秒
        .setRetryPolicy(FixedDelayRetryPolicy.builder()
                .attempts(5)
                .delay(Duration.ofSeconds(2))
                .build())
        // 可选：设置批次大小和间隔，匹配你的业务吞吐需求
        .setBatchSize(1000)
        .setBatchInterval(Duration.ofSeconds(1))
        .build();

额外注意事项 #

• 必须开启Flink检查点：精确一次语义依赖检查点，你需要在Flink作业中配置检查点，比如：

  env.enableCheckpointing(30000); // 每30秒触发一次检查点
  env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
  

• Kinesis流的分片配置：确保每个key对应的分片有足够的吞吐量，减少超限错误的发生，从源头降低重试场景。
• 故障转移后的恢复：当重试耗尽任务失败时，Flink会从最近的检查点恢复，重新处理失败的批次，此时不会有新批次的记录混入，保证批次顺序。

这种配置下，批次1会被完全处理（成功写入或故障恢复后写入），之后才会开始处理批次2，完全符合你要求的批次间顺序；同时同一key的记录会被分配到同一Flink算子实例，按顺序写入对应Kinesis分片，保证key级别的顺序。

内容的提问来源于stack exchange，提问作者Carlos Botelho