Kafka 与 Flink 流批一体集成

Kafka：数据总线（Buffer of Record）
  - 承接上游写入，缓冲削峰
  - 多消费者复用（回放、审计、下游）
  - 分区并行、持久化

Flink：流式计算引擎（Stateful Stream Processing）
  - 有状态算子（聚合、Join、CEP）
  - 事件时间 + 水位线（乱序处理）
  - 精确一次（checkpoint + 两阶段提交）

一个 Kafka 分区 → 最多被一个 Source 子任务消费（保证顺序）
Source 并行度 > 分区数 → 部分子任务空闲（浪费）
Source 并行度 < 分区数 → 部分子任务消费多分区

KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers("kafka:9092")
    .setTopics("orders")
    .setGroupId("flink-orders")             // 注意：Flink 用 group 管理 offset
    .setStartingOffsets(OffsetsInitializer.committedOffsets(
        OffsetResetStrategy.EARLIEST))
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

DataStream<String> stream = env.fromSource(
    source,
    WatermarkStrategy.noWatermarks(),
    "kafka-source"
);

// 生产推荐：优先用已提交 offset，无则回退 earliest
.setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))

.setProperty("partition.discovery.interval.ms", "60000")   // 每 60s 发现新分区

① JobManager 触发 checkpoint barrier
② barrier 随数据流向下游传播
③ 每个算子将状态写入 state backend（RocksDB/HDFS）
④ Source 记录当前消费的 offset
⑤ 所有算子确认 → checkpoint 完成（全局一致点）

// 默认：checkpoint 成功后才提交 offset 到 Kafka
// 这是「不丢不重」的关键——避免 offset 提交超前于状态
env.enableCheckpointing(60000);   // 60s 一次

// 若开启「提交 offset 到 Kafka」（供外部监控用）
.setProperty("commit.offsets.on.checkpoint", "true");  // 默认即 true

At-most-once：不启 checkpoint → 故障后从 Kafka 最新位点消费（可能丢）
At-least-once：启 checkpoint，Sink 非事务 → 故障后重放（可能重）
Exactly-once：启 checkpoint + 事务 Sink → 不丢不重

① 预提交（Pre-commit）：checkpoint 期间，把数据写入 Kafka 事务（未提交）
② checkpoint 完成：Flink 通知所有算子 checkpoint 成功
③ 提交（Commit）：提交 Kafka 事务，数据对下游可见

KafkaSink<String> sink = KafkaSink.<String>builder()
    .setBootstrapServers("kafka:9092")
    .setRecordSerializer(KafkaRecordSerializationSchema.builder()
        .setTopic("orders-result")
        .setValueSerializationSchema(new SimpleStringSchema())
        .build())
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)   // 关键
    .setTransactionalIdPrefix("flink-orders-sink")           // 事务前缀
    .setProperty("transaction.timeout.ms", "900000")         // 事务超时
    .build();

transaction.timeout.ms 必须 > checkpoint 间隔 + 恢复时间
否则：事务超时被 broker 中止 → 数据丢失
Kafka broker 的 transaction.max.timeout.ms 默认 15 分钟 → 上限

Source：可回放（Kafka offset） ✓
算子：checkpoint 一致 ✓
Sink：事务或幂等（Kafka 事务 / 幂等写 / Upsert） ✓

流模式：Kafka Source → 无界流 → 持续处理
批模式：Kafka Source（有界，读到最新 offset 停止）→ 有界流 → 批处理

.setBounded(OffsetsInitializer.latest())   // 读到当前最新位点即停止 → 批模式

// 用执行模式切换流/批，代码不变
env.setRuntimeMode(RuntimeExecutionMode.STREAMING);  // 或 BATCH

同一逻辑：实时链路（流）+ 回补链路（批）用同一份 SQL/代码
避免「实时一套、离线一套」的双份维护与口径不一致

// 从 Kafka 消息中提取事件时间，生成水位线
WatermarkStrategy<Order> wm = WatermarkStrategy
    .<Order>forBoundedOutOfOrderness(Duration.ofSeconds(5))   // 容忍 5s 乱序
    .withTimestampAssigner((o, ts) -> o.getEventTime())
    .withIdleness(Duration.ofMinutes(1));                     // 空闲分区处理

env.fromSource(source, wm, "kafka-source");

水位线延迟越大 → 容忍乱序越多 → 窗口触发越晚（结果延迟）
水位线延迟越小 → 结果越快 → 迟到数据越多（被丢弃或侧输出）

// 迟到数据侧输出，而非直接丢弃
OutputTag<Order> lateTag = new OutputTag<>("late-orders"){};
SingleOutputStreamOperator<Result> result = stream
    .windowAll(TumblingEventTimeWindows.of(Time.minutes(1)))
    .allowedLateness(Time.seconds(30))       // 允许 30s 迟到
    .sideOutputLateData(lateTag)
    .apply(agg);

水位线 = min(所有输入分区的水位线)
→ 某个分区长时间无数据（空闲）会拖住全局水位线
→ 用 withIdleness 标记空闲分区，跳过其水位线

// 消费端（Flink Kafka Source）
"partition.discovery.interval.ms" = "60000"
"fetch.min.bytes" = "1"
"max.partition.fetch.bytes" = "1048576"

// checkpoint
env.enableCheckpointing(60000);
env.getCheckpointConfig().setCheckpointTimeout(600000);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(30000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);

Kafka 生产速率 > Flink 处理速率 → Source 被反压 → Lag 上涨
Flink Web UI 的 backpressure 面板可定位瓶颈算子
对策：扩容并行度、优化算子、增大 checkpoint 容忍