端到端一致性的三层模型
eventd v3 的数据流涉及三个外部系统：
Kafka (eventd-raw) → Kafka Streams → Kafka (eventd-cleaned)
│
↓
Flink DataStream → Redis
端到端一致性涉及三层：
| 层 | 系统 | 一致性机制 |
|---|---|---|
| Source | Kafka | offset 随 Checkpoint 提交（可重放） |
| Internal | Flink | Checkpoint 一致快照（barrier 对齐） |
| Sink | Redis | 幂等写入（确定性 key） |
三层的机制不是同一回事：Source 靠「可重放的位点」，Flink 内部靠「一致快照回滚」，Sink 靠「重复写不出错」。端到端 EOS 是由这三件事拼出来的，而不是某处配一个开关。
Kafka Source 的 Exactly-Once
KafkaSource 没有 EXACTLY_ONCE 这样的模式参数。它的语义由两件事决定：offset 只在 Checkpoint 完成时提交；消费位置能从 Checkpoint 恢复并重读。
KafkaSource<Event> source = KafkaSource.<Event>builder()
.setBootstrapServers("kafka:9092")
.setTopics("eventd-cleaned")
.setGroupId("eventd-analytics")
.setStartingOffsets(OffsetsInitializer.earliest())
// 不设反序列化器时 build() 会直接抛异常；定义见 7.1
.setValueOnlyDeserializer(new EventDeserializationSchema())
// 默认 true：offset 随 Checkpoint 提交；设成 false 就只能靠 group 自动提交，
// 那样 offset 与状态快照对不上，恢复会丢或重
.setProperty("commit.offsets.on.checkpoint", "true")
// 上游是 Kafka Streams EOS 写出的事务消息时才需要；普通 topic 加不加都一样
.setProperty("isolation.level", "read_committed")
.build();
Kafka Streams 的 EOS 配置（eventd-cleansing）：
props.put(StreamsConfig.PROCESSING_GUARANTEE_CONFIG, StreamsConfig.EXACTLY_ONCE_V2);
这意味着：
- Kafka Streams 处理完一批数据后，原子性提交 offset 和生产输出
- 消费者只读取已提交的事务消息
- 故障时重放：从最后提交的 offset 开始，确保不丢不重
Checkpoint：内部一致快照
Flink 内部的 EOS 来自 Chandy-Lamport 式的 barrier 对齐，不是 2PC：
JobManager 周期性插入 barrier
Source: 快照 offset，barrier 顺流下发
Map/Window: 收到 barrier 时快照本地状态，向下转发
Sink: 收到 barrier 时快照自身状态
全部子任务确认 → Checkpoint 完成 → Source 提交这批 offset
任一步骤失败 → 本次 Checkpoint 丢弃 → 作业回滚到最后一次完成的快照
注意 Sink 在这里只快照自己的状态，并不保证对外部系统写入的原子性。外部系统的语义由 Sink 自己决定，见下一节。
只有在 Sink 实现了 TwoPhaseCommitSinkFunction（或 Sink V2 的预提交协议，如 KafkaSink 的 DeliveryGuarantee.EXACTLY_ONCE）时，才会出现真正的事务两阶段提交：barrier 到达时开启事务并缓存写入，Checkpoint 完成时提交事务，作业回滚时丢弃（abort）未提交事务。Redis 没有这种 Sink，所以 eventd v3 的 Redis 侧只能走幂等。
配置：
env.enableCheckpointing(30000);
env.getCheckpointConfig().setCheckpointingConsistencyMode(
org.apache.flink.core.execution.CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setMinPauseBetweenCheckpoints(10000);
Redis Sink 的幂等写入
Redis 没有事务型 Sink，eventd v3 用确定性 key + 覆盖写让重复写无害：
public class IdempotentRedisSink
extends RichSinkFunction<AnalyticsJob.MetricRecord> {
// 字段名与 9.3 的 MetricRecord(metric, dimension, dimensionValue,
// windowStart, windowEnd, payload) 保持一致
private transient RedisClient client;
private transient StatefulRedisConnection<String, String> connection;
private transient RedisCommands<String, String> sync;
@Override
public void open(OpenContext openContext) {
client = RedisClient.create("redis://redis:6379");
connection = client.connect();
sync = connection.sync();
}
@Override
public void invoke(AnalyticsJob.MetricRecord record, Context context) {
// key 由 (指标, 窗口起点, 维度值) 唯一确定：同一窗口重放 → 同一个 key
String key = String.format("eventd:metrics:%s:%d:%s",
record.metric(), record.windowStart(), record.dimensionValue());
sync.setex(key, 300, record.payload());
}
@Override
public void close() {
if (connection != null) {
connection.close();
}
if (client != null) {
client.shutdown();
}
}
}
为什么 SET 是幂等的？
因为 key 与 value 都由窗口内容唯一确定：同一个窗口重放时算出同一个 key、同一个值，覆盖写不会留下第二条记录。这依赖两个前提，缺一不可：
- 窗口结果是确定性的。用了
now()、随机数、当前汇率等运行时输入，重算结果就会不同。 - 输出只按窗口键落盘，不追加到列表。
「latest 指针」不满足这两条，所以不要天真地覆盖：
// 反例：重放旧窗口时会把 latest 指针写回过去
sync.set("eventd:metrics:" + record.metric() + ":latest", record.payload());
要让 latest 也幂等，需要在写之前比较窗口起点。payload 里只有聚合值（如 {"count":42}），没有窗口起点，所以把窗口起点单独存一个键，再用 Lua 把「比较 + 写」做成原子操作：
-- KEYS[1] = latest payload，KEYS[2] = latest 的 windowStart
-- ARGV[1] = 本次 windowStart，ARGV[2] = 本次 payload
local cur = redis.call('GET', KEYS[2])
if cur == false or tonumber(cur) <= tonumber(ARGV[1]) then
redis.call('SET', KEYS[1], ARGV[2])
redis.call('SET', KEYS[2], ARGV[1])
return 1
end
return 0
sync.eval(
"local cur = redis.call('GET', KEYS[2]) " +
"if cur == false or tonumber(cur) <= tonumber(ARGV[1]) then " +
" redis.call('SET', KEYS[1], ARGV[2]) " +
" redis.call('SET', KEYS[2], ARGV[1]) return 1 end return 0",
ScriptOutputType.INTEGER,
new String[]{
"eventd:metrics:" + record.metric() + ":latest",
"eventd:metrics:" + record.metric() + ":latest:windowStart"},
String.valueOf(record.windowStart()),
record.payload());
两个键在同一个 Lua 脚本里读写，Redis 单线程执行脚本，所以不会出现「payload 更新了但 windowStart 没更新」的中间态。
读到与判定的原子性由单条 Lua 保证；跨多个 key 的更新仍然不是事务，Redis 没有跨 key 的回滚。
幂等性设计原则
eventd v3 各组件的幂等策略：
| 组件 | 操作 | 幂等策略 |
|---|---|---|
| Kafka Streams 去重 | put(eventId, now) | key = eventId（唯一）；value 是首次见到的时间，供 TTL 清理 |
| Flink 窗口聚合 | 增量累加 | 状态在 Checkpoint 里，恢复后从快照继续 |
| Redis Sink | SETEX key | key 由窗口内容确定，重复覆盖无害 |
故障恢复能测到什么
在本地 createLocalEnvironment 里跑「抛异常 → 期待自动重启」是测不出 EOS 的：
- 本地执行器的异常会一路抛到
execute()，作业直接失败退出； - 即便用 MiniCluster，重启也要靠
restart-strategy生效并重新从 Checkpoint 拉取 Source，而集合源（fromCollection）没有 offset 状态，重放行为与 Kafka 完全不同； - 最关键的是，测试里必须有一个能被重复写入并检查副作用的外部系统，否则即使真的重放了，你也看不到「重复」。
所以 eventd v3 的恢复测试分两层做：
| 层 | 怎么测 | 能证明什么 |
|---|---|---|
| 幂等 key 单元测试 | 对同一 MetricRecord 调两次 keyOf(record)，断言 key 相等 | key 设计确定，覆盖写是无害的 |
| 端到端重放测试 | Testcontainers 起 Kafka + Redis，作业跑完 Checkpoint 后 kill TaskManager，重启后统计 Redis 里的键数 | 重放之后没有多余键 |
第一层是纯函数测试，跑得飞快：
// 与 IdempotentRedisSink.invoke 里的格式化规则保持一致；
// 生产代码可以把这段抽成 MetricRecord 的静态方法，测试直接复用
private static String keyOf(MetricRecord record) {
return String.format("eventd:metrics:%s:%d:%s",
record.metric(), record.windowStart(), record.dimensionValue());
}
@Test
void sameWindowProducesSameKey() {
// 两个内容相同但实例不同的记录，模拟「重放算出同一个窗口」
MetricRecord first = new MetricRecord(
"order-count", "symbol", "BTC", 1_700_000_000_000L,
1_700_000_060_000L, "{\"count\":42}");
MetricRecord replay = new MetricRecord(
"order-count", "symbol", "BTC", 1_700_000_000_000L,
1_700_000_060_000L, "{\"count\":42}");
assertEquals(keyOf(first), keyOf(replay));
assertEquals("eventd:metrics:order-count:1700000000000:BTC", keyOf(first));
}
第二层需要真实集群和真实外部系统，才能观察「重复写到底有没有副作用」。它的断言不是「作业没报错」，而是「Redis 里键的数量等于唯一窗口数」——重复写不会增加键数，这才是幂等生效的证据。这类测试放在 CI 的集成阶段，而不是每个 mvn test 都跑。
端到端一致性检查清单
□ Kafka Source: commit.offsets.on.checkpoint=true（默认），确有必要才加 read_committed
□ Kafka Streams: EXACTLY_ONCE_V2 配置
□ Flink Checkpoint: EXACTLY_ONCE 模式 + 合理间隔
□ Flink RestartStrategy: 自动从 Checkpoint 恢复
□ Redis Sink: 幂等 key 设计（确定性 key）
□ 窗口计算: 确定性（相同输入 → 相同输出）
□ 去重逻辑: 基于 eventId（唯一标识）
语义降级
某些场景下可以接受 AT_LEAST_ONCE：
| 场景 | 推荐语义 | 原因 |
|---|---|---|
| 实时面板指标 | AT_LEAST_ONCE | 偶尔重复不影响趋势判断 |
| 告警触发 | EXACTLY_ONCE | 重复告警会干扰 |
| 计费统计 | EXACTLY_ONCE | 不能重复计费 |
| 日志聚合 | AT_LEAST_ONCE | 重复日志可去重后处理 |
降级配置：
// AT_LEAST_ONCE：性能更好，但不保证不重复
env.getCheckpointConfig().setCheckpointingConsistencyMode(
org.apache.flink.core.execution.CheckpointingMode.AT_LEAST_ONCE);
小结
| 组件 | 一致性 | 机制 |
|---|---|---|
| Kafka Source | 可重放 | offset 随 Checkpoint 提交 + read_committed |
| Flink 内部 | EXACTLY_ONCE | barrier 对齐的一致快照 |
| Redis Sink | 幂等 | 确定性 key + 覆盖写 |
| 整体 | 端到端 EOS（有前提） | 三层叠加，且窗口结果必须确定 |
最后一行的「有前提」不能省：Redis 侧没有任何回滚机制，所谓端到端 EOS 实际是「重放不会改变最终结果」。一旦窗口结果依赖运行时不变量以外的东西，或者用了 latest 这类非确定性键，语义就退化成了至少一次。
下一节 11.2 多流 Join 与关联分析 ：Interval Join、CoProcessFunction。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。