流处理集群的元数据一致性：ZooKeeper 到基于 Raft 自实现元数据服务的迁移

一、ZK 在流处理集群中的三个痛点

流处理系统（Flink、Kafka Streams）依赖 ZooKeeper 管理集群元数据——包括 TaskManager 注册、Checkpoint 路径、JobGraph 状态。ZooKeeper 在核心场景下表现稳定，但流处理集群的规模从 50 节点增长到 500 节点后，以下问题变得不可忽视：

Watch 风暴：所有 TaskManager 在 ZK 创建 Ephemeral Node 以注册存活状态。ZK Session 过期时（如网络抖动），数百个 Ephemeral Node 被同时删除，触发数千个 Watch 事件。ZK 处理这些事件期间，集群不可用——这就是著名的"herd effect"（羊群效应）。
运维复杂度：ZK 是独立的 Java 进程，需要单独部署、监控、升级。它与流处理引擎的技术栈不同（Java vs Rust），运维团队需要掌握两套工具。
性能天花板：ZK 的写操作需要半数以上节点确认（ZAB 协议）。在高频元数据更新（如 TaskManager 每 5 秒上报心跳指标）场景下，ZK 的写入吞吐受限于单 Leader 的处理能力。

自实现基于 Raft 的元数据服务的动机不是"造轮子"，而是消除外部依赖、获取对一致性协议的完全控制权。当需要定制如"Checkpoint 元数据按租户隔离"、"TaskManager 心跳的批量确认"等特性时，外部系统无法提供所需的灵活性。

二、元数据服务的架构迁移

迁移的核心工作是将 ZK 的两种核心原语映射到 Raft 实现：

ZK Path → Raft KV 存储：ZK 的层级路径（/flink/taskmanagers/tm-1）映射为 Raft 状态机中的 KV 对（key="taskmanagers/tm-1"）。创建、读取、更新、删除（CRUD）操作通过 Raft 的日志复制实现。

ZK Ephemeral Node → Raft Lease：ZK 的临时节点与 Client Session 绑定——Session 断开后自动删除。Raft 中没有等同概念，需要实现租约（Lease）机制：客户端定期发送心跳续约，Raft Leader 维护租约 TTL。TTL 过期后自动删除该客户端注册的所有临时数据。

ZK Watch → Raft 事件通知：ZK 的 Watch 允许客户端订阅某节点的变更事件。在 Raft 中，状态机的每次 Apply 可以触发回调——通知订阅了对应 key 的客户端。

三、嵌入式 Raft 元数据服务的 Rust 实现

use std::collections::{HashMap, BTreeMap};
use std::sync::Arc;
use tokio::sync::{RwLock, Mutex, mpsc};
use serde::{Serialize, Deserialize};
use chrono::{Utc, Duration};
/// 元数据操作类型
#[derive(Clone, Serialize, Deserialize, Debug)]
pub enum MetaOperation {
    /// 创建/更新 KV
    Put { key: String, value: Vec<u8>, ephemeral: bool },
    /// 删除 KV
    Delete { key: String },
    /// CAS 操作: 防空创建
    CreateIfAbsent { key: String, value: Vec<u8>, ephemeral: bool },
    /// 租约续约
    RenewLease { client_id: String },
}
/// 元数据条目
#[derive(Clone, Debug)]
pub struct MetaEntry {
    pub value: Vec<u8>,
    /// 是否为临时节点（绑定到租约）
    pub ephemeral: bool,
    /// 所属客户端 ID（仅 ephemeral 节点有效）
    pub owner: Option<String>,
    /// 版本号 —— 用于 CAS 操作
    pub version: u64,
    /// 创建时间
    pub created_at: chrono::DateTime<Utc>,
}
/// Raft 状态机 —— 存储元数据
pub struct MetadataStateMachine {
    /// KV 存储
    kv: BTreeMap<String, MetaEntry>,
    /// 租约管理: client_id → 过期时间
    leases: HashMap<String, chrono::DateTime<Utc>>,
    /// 租约 TTL（秒）
    lease_ttl: i64,
    /// Watch 订阅者: key_prefix → 通知通道列表
    watchers: HashMap<String, Vec<mpsc::UnboundedSender<WatchEvent>>>,
}
/// Watch 事件
#[derive(Clone, Debug)]
pub struct WatchEvent {
    pub key: String,
    pub event_type: WatchEventType,
    pub value: Option<Vec<u8>>,
}
#[derive(Clone, Debug)]
pub enum WatchEventType {
    Created,
    Updated,
    Deleted,
}
impl MetadataStateMachine {
    pub fn new(lease_ttl_secs: i64) -> Self {
        Self {
            kv: BTreeMap::new(),
            leases: HashMap::new(),
            lease_ttl: lease_ttl_secs,
            watchers: HashMap::new(),
        }
    }
    /// 应用一个操作到状态机
    /// 
    /// 关键设计：所有写操作通过此方法执行，
    /// 由 Raft 的 Apply 循环调用。这保证了状态机的变更是确定性的。
    pub fn apply(&mut self, op: &MetaOperation) -> Result<Option<Vec<u8>>, MetaError> {
        match op {
            MetaOperation::Put { key, value, ephemeral } => {
                let now = Utc::now();
                let entry = MetaEntry {
                    value: value.clone(),
                    ephemeral: *ephemeral,
                    owner: None, // Put 操作不关联 owner（CreateIfAbsent 才关联）
                    version: 0,
                    created_at: now,
                };
                let old = self.kv.insert(key.clone(), entry);
                // 通知 Watch 订阅者
                self.notify_watchers(key, if old.is_some() {
                    WatchEventType::Updated
                } else {
                    WatchEventType::Created
                }, Some(value.clone()));
                Ok(old.map(|e| e.value))
            }
            MetaOperation::Delete { key } => {
                let old = self.kv.remove(key);
                if let Some(entry) = &old {
                    self.notify_watchers(key, WatchEventType::Deleted, Some(entry.value.clone()));
                }
                Ok(old.map(|e| e.value))
            }
            MetaOperation::CreateIfAbsent { key, value, ephemeral } => {
                if self.kv.contains_key(key) {
                    return Err(MetaError::AlreadyExists);
                }
                // 等价于 Put，但只有 key 不存在时才执行
                self.apply(&MetaOperation::Put {
                    key: key.clone(),
                    value: value.clone(),
                    ephemeral: *ephemeral,
                })
            }
            MetaOperation::RenewLease { client_id } => {
                // 更新租约过期时间
                // 心跳 = client_id 的过期时间推迟 lease_ttl 秒
                let expiry = Utc::now() + Duration::seconds(self.lease_ttl);
                self.leases.insert(client_id.clone(), expiry);
                Ok(None)
            }
        }
    }
    /// 租约 GC —— 定期清理过期的临时节点
    /// 
    /// 应在独立的后台 Task 中定期运行（如每 1 秒）
    pub fn gc_expired_leases(&mut self) -> Vec<String> {
        let now = Utc::now();
        let mut expired_clients = Vec::new();
        // 1. 收集过期的租约
        for (client_id, expiry) in &self.leases {
            if *expiry < now {
                expired_clients.push(client_id.clone());
            }
        }
        // 2. 删除过期客户端的临时节点
        let mut keys_to_delete = Vec::new();
        for (key, entry) in &self.kv {
            if entry.ephemeral {
                if let Some(owner) = &entry.owner {
                    if expired_clients.contains(owner) {
                        keys_to_delete.push(key.clone());
                    }
                }
            }
        }
        // 3. 删除临时节点和租约记录
        for key in &keys_to_delete {
            self.kv.remove(key);
            self.notify_watchers(key, WatchEventType::Deleted, None);
        }
        for client_id in &expired_clients {
            self.leases.remove(client_id);
        }
        keys_to_delete
    }
    /// 注册 Watch —— 订阅特定 key 前缀的变更事件
    pub fn watch(&mut self, key_prefix: &str, tx: mpsc::UnboundedSender<WatchEvent>) {
        self.watchers.entry(key_prefix.to_string())
            .or_insert_with(Vec::new)
            .push(tx);
    }
    /// 通知所有匹配的 Watcher
    fn notify_watchers(&self, key: &str, event_type: WatchEventType, value: Option<Vec<u8>>) {
        let event = WatchEvent {
            key: key.to_string(),
            event_type,
            value,
        };
        for (prefix, senders) in &self.watchers {
            if key.starts_with(prefix) {
                for tx in senders {
                    let _ = tx.send(event.clone());
                }
            }
        }
    }
}
/// Raft 元数据服务的客户端 SDK
pub struct MetadataClient {
    /// 向 Raft Leader 发送操作的通道
    propose_tx: mpsc::UnboundedSender<MetaOperation>,
    /// 租约心跳间隔
    heartbeat_interval: std::time::Duration,
    /// 客户端 ID
    client_id: String,
}
impl MetadataClient {
    /// 注册 TaskManager —— 使用 CreateIfAbsent 保证唯一性
    pub async fn register_taskmanager(
        &self,
        tm_id: &str,
        address: &str,
    ) -> Result<(), MetaError> {
        let key = format!("taskmanagers/{}", tm_id);
        let value = serde_json::to_vec(&serde_json::json!({
            "address": address,
            "registered_at": Utc::now().to_rfc3339(),
        }))?;
        // 使用 CreateIfAbsent —— 如果 TM 已注册，返回 AlreadyExists
        // 防止网络重试导致的重复注册
        let op = MetaOperation::CreateIfAbsent {
            key,
            value,
            ephemeral: true, // 临时节点：客户端断开后自动删除
        };
        self.propose_tx.send(op)
            .map_err(|_| MetaError::ChannelClosed)?;
        Ok(())
    }
    /// 启动租约心跳循环
    /// 
    /// 心跳间隔 = TTL / 3（确保在 TTL 过期前至少续约 2 次）
    pub async fn start_heartbeat(&self) {
        let tx = self.propose_tx.clone();
        let client_id = self.client_id.clone();
        let interval = self.heartbeat_interval;
        tokio::spawn(async move {
            loop {
                let _ = tx.send(MetaOperation::RenewLease {
                    client_id: client_id.clone(),
                });
                tokio::time::sleep(interval).await;
            }
        });
    }
    /// 更新 TaskManager 心跳指标
    pub async fn report_metrics(
        &self,
        tm_id: &str,
        metrics: &HashMap<String, f64>,
    ) -> Result<(), MetaError> {
        let key = format!("taskmanagers/{}/metrics", tm_id);
        let value = serde_json::to_vec(metrics)?;
        self.propose_tx.send(MetaOperation::Put {
            key,
            value,
            ephemeral: false, // 持久化节点：心跳指标在 TM 断开后保留
        }).map_err(|_| MetaError::ChannelClosed)?;
        Ok(())
    }
}
#[derive(Debug)]
pub enum MetaError {
    AlreadyExists,
    ChannelClosed,
    Serialize(serde_json::Error),
}
impl From<serde_json::Error> for MetaError {
    fn from(e: serde_json::Error) -> Self { MetaError::Serialize(e) }
}

关键设计决策：

CreateIfAbsent 操作：这是 ZK 的 CreateMode.PERSISTENT 在 ZK 中的等价操作。在 Raft 状态机中实现 CAS（Compare-And-Swap）语义，防止并发注册导致的数据覆盖。
心跳间隔 = TTL / 3：如果 TTL = 30 秒，心跳每 10 秒发送一次。这样即使某次心跳因网络丢包而丢失，仍有两次续约机会。
BTreeMap 而非 HashMap：元数据量通常不大（< 10000 条目），但需要支持范围查询（如"所有 taskmanagers/ 下的节点"）。BTreeMap 的 range 方法提供 O(log n + k) 的前缀查询，这是 ZK 的 getChildren 的等价操作。
Watch 通知使用 UnboundedSender：避免 Watch 回调阻塞状态机的 Apply 过程。但如果客户端消费不过来，Unbounded 通道会无限增长——生产环境应使用 Bounded 通道 + Drop Old 策略。

四、元数据服务迁移的适用边界与权衡

适用场景：

已有 Raft 基础库（或 Rust 技术栈），需要一个轻量级的嵌入式元数据存储。
元数据读写频率高（> 1000 ops/s），ZK 成为瓶颈。
需要定制化功能（如元数据分片、租户级隔离），外部系统无法满足。

不适用场景：

集群规模小（< 10 节点），ZK 的运维成本远低于自建服务。
需要与其他系统（Kafka、HBase）共享元数据——ZK 的通用性在此是优势。
团队没有 Raft 实现和运维经验——自建分布式共识系统的 Bug 代价极高。

主要权衡：

嵌入式 vs Sidecar：将 Raft 节点嵌入流处理进程消除了网络通信开销，但可能导致流处理 Heap 被元数据占用。Sidecar 模式隔离资源但增加了一层网络跳转。
Raft 日志大小：高频的指标上报（每 5 秒一次，500 个 TM）会导致 Raft 日志快速增长。需要定期做快照（Snapshot）压缩日志。
Watch 机制的语义保证：ZK 的 Watch 是一次性的（触发后需要重新注册），这是故意设计——迫使客户端在事件处理后重新读取最新状态。Raft 的 Watch 可设计为持久性订阅，但需要客户端自己处理事件丢失。