Hi all, I would like to start a discussion for FIP-52: ZooKeeper Removal and Embedded Raft Metadata Backend:
https://cwiki.apache.org/confluence/spaces/FLUSS/pages/449282216/FIP-52+ZooKeeper+Removal+and+Embedded+Raft+Metadata+Backend Issue: https://github.com/apache/fluss/issues/4014 Branch: feature/zookeeper-removal-raft Please keep the discussion on this list rather than commenting on the wiki. ---------------------------------------------------------------------- Why ---------------------------------------------------------------------- Fluss today depends on an external ZooKeeper ensemble for cluster coordination, metadata persistence, sequence IDs, TabletServer liveness (ephemeral znodes), dynamic config, ACLs, and related leases. Clients never talk to ZooKeeper (bootstrap.servers + Admin RPC only); servers do. Operators still have to run ZooKeeper for every Fluss cluster: - A production cluster cannot start without a healthy ZK ensemble (quorum, tick time, session timeout, SASL/ACL, backups). - ZK issues cascade into coordinator leadership, TabletServer registration, and metadata availability. - Extra network hops and a second failure domain sit on the metadata path. - Container, Kubernetes, edge, and single-node developer setups pay for an extra stateful service that exists only for coordination. This is already the published architecture direction: ZooKeeper is a transitional dependency, to be replaced by KvStore for metadata and Raft for coordination. ---------------------------------------------------------------------- What this FIP proposes ---------------------------------------------------------------------- Replace ZooKeeper with an embedded Apache Ratis Raft quorum on CoordinatorServers, plus a RocksDB-backed metadata KvStore. Only CoordinatorServers are Raft voters. TabletServers are observers: they discover a coordinator from coordinator.seed.nodes (optionally expanded by gossip), register/heartbeat over RPC, and consume a pull-based metadata stream. Today Target ----- ------ Coordinator ──► ZooKeeper Coordinator quorum TabletServer ──► ZooKeeper └── embedded Ratis + KvStore Clients ──► bootstrap.servers TabletServer ──► Register / Heartbeat / FetchMetadata Clients ──► bootstrap.servers (unchanged) Consensus engine is Apache Ratis. Metadata apply is KRaft-style commit-apply: every node, including the leader, mutates the store only after a log entry is committed. Leader-only serving for strongly consistent reads; followers redirect. The Raft term is the coordinator epoch. How each ZooKeeper role is replaced: - Coordinator leader election (Curator latch) -> Ratis election + RaftLeaderSelector (gainPrimacy / losePrimacy, catch-up before serving) - Database / table / partition / schema / assignment / LeaderAndIsr / ACL / config / leases -> typed MetadataRecords in the Raft log, applied into a RocksDB-backed metadata KvStore (same JSON serdes as ZK values) - Sequence IDs -> RaftSequenceIDCounter with pre-reserved blocks (coordinator.raft.id-block-size) - TabletServer ephemeral znode + session -> RegisterTabletServer + heartbeat lease; timeout appends a replicated FENCE_TABLET_SERVER - ZK watches -> in-process watches on the leader store; cross-process FetchMetadata stream (offset = Raft index) - zkCli observability -> fluss raft status / leader + Raft metrics We borrow KRaft patterns (commit-apply, record log, broker-style registration/heartbeat, observer tablets, id blocks, metadata stream) and reject KRaft-the-codebase. Fluss needs an embeddable Java Raft library with a pluggable state machine; that is Apache Ratis, shaded as fluss-shaded-ratis (org.apache.fluss.shaded.ratis.*). ---------------------------------------------------------------------- What we are asking the community to accept ---------------------------------------------------------------------- This FIP asks the community to accept Phases 1-5 as the merge/release shape: 1. Dual-support (ZK or Raft via config) 2. Testing (dual-backend HA, Raft correctness, chaos ITs) 3. Migration tools 4. Docs 5. Raft as the default Phase 6 (delete ZooKeeper mode and the shaded ZK dependency) is a future major and is out of the first merge. That deletion is not part of the first release of this work. Goals for this work: 1. Default deploy is Fluss-only: CoordinatorServers form a Raft group; no ZooKeeper process is required. 2. A supported dual-backend window so existing ZK clusters can keep running and migrate with tools, not a flag day. 3. No change to client APIs, log/KV format, or bootstrap.servers. 4. A later major (Phase 6) removes ZK mode after soak. ---------------------------------------------------------------------- Compatibility (clients do not change) ---------------------------------------------------------------------- Unchanged: - Client bootstrap is still bootstrap.servers. fluss-client has no ZooKeeper dependency. - Admin / DDL / produce / fetch / lookup public APIs and table RPCs. - Binary log / KV format. - Connector / SDK wire protocol. Downstream *test harnesses* that launch a ZK container and inject zookeeper.address must switch to seed nodes when the server image defaults to Raft. Existing ZK clusters keep working if zookeeper.address remains set and coordinator.seed.nodes is not set. Mixed ZK+Raft in one cluster is unsupported (no dual-write). ---------------------------------------------------------------------- Backend selection and configuration ---------------------------------------------------------------------- Backend selection is exclusive. Setting both keys fails fast (IllegalConfigurationException): 1. zookeeper.address only -> ZOOKEEPER (deprecated; warn at startup) 2. coordinator.seed.nodes only -> RAFT 3. Both -> reject 4. Neither -> RAFT (current default on the implementation branch) A silent default flip to Raft on upgrade would be a foot-gun; existing ZK clusters MUST keep an explicit zookeeper.address on upgrade. Docs and a startup warning exist for that reason. New Raft / discovery keys (defaults in ConfigOptions): coordinator.raft.voters static voter set host:raftPort coordinator.raft.node.id this node's Raft peer id coordinator.raft.port 9124 (dedicated Raft gRPC; not client-facing) coordinator.raft.dir local Raft log + metadata KvStore directory coordinator.raft.heartbeat.interval 500ms coordinator.raft.election.timeout 1s coordinator.raft.first-election.timeout 300ms coordinator.raft.catchup.timeout 30s (new leader must apply through leadership-gain index before serving) coordinator.raft.log.segment-size 8mb coordinator.raft.snapshot.trigger.threshold 10000 coordinator.raft.id-block-size 1000 coordinator.tablet-server.session-timeout 30s coordinator.metadata-stream.max-wait 500ms coordinator.gossip.enabled false (optional SWIM-style TCP gossip to expand a *partial* seed list; does not replace raft.voters) coordinator.gossip.port 9125 coordinator.raft.tls.enabled false Firewall coordinator.raft.port (and gossip, if enabled) to coordinator hosts. They are not client ports. Deprecated but kept until Phase 6: zookeeper.address, zookeeper.path.root, zookeeper.client.*. Gossip is not consensus. Static seeds still select the Raft backend. ---------------------------------------------------------------------- New RPCs, CLI, and ops surface ---------------------------------------------------------------------- New PRIVATE / PUBLIC RPCs in FlussApi.proto / ApiKeys. Mixed-version clients that only use existing Admin/data APIs are unaffected. FETCH_METADATA 1065 PRIVATE pull committed metadata records by offset FETCH_METADATA_SNAPSHOT 1066 PRIVATE bootstrap when from_offset was compacted REGISTER_TABLET_SERVER 1067 PRIVATE replaces ephemeral /tabletservers/ids/[id] TABLET_SERVER_HEARTBEAT 1068 PRIVATE in-memory lease renewal; expiry appends fence ALLOCATE_ID 1069 PRIVATE contiguous ID range from a Raft sequence counter GET_COORDINATOR_RAFT_STATUS 1070 PUBLIC leader identity + voter lag ADD_RAFT_VOTER 1071 PUBLIC quorum add; leader-only REMOVE_RAFT_VOTER 1072 PUBLIC quorum remove with safety checks On the ZooKeeper backend, the PRIVATE TabletServer-facing RPCs are not served (tablets still register via ZK). New bin/fluss umbrella (FlussCli): fluss raft bootstrap fluss raft status [--watch] [--interval ms] [--json] fluss raft leader fluss raft add-voter / remove-voter fluss raft unsafe-recover --i-understand-data-loss fluss zk-to-raft-export fluss zk-to-raft-import fluss zk-to-raft-validate fluss zk-to-raft-live-migrate unsafe-recover is offline, majority-loss recovery on the most up-to-date survivor only; it can lose unreplicated commits. It never runs automatically. Coordinators expose Raft gauges on the existing fluss-metrics MetricGroup (raftRole, raftTerm, raftCommitIndex, raftAppliedIndex, raftApplyLag, raftFollowerLag). No new public Java API for end users. ---------------------------------------------------------------------- Migration ---------------------------------------------------------------------- Recommended path (also documented in website/docs/install-deploy/migrate-zk-to-raft.md): 1. Export ZK (zk-to-raft-export). First export may run online. 2. Provision an empty Raft quorum; fluss raft bootstrap. 3. Import (zk-to-raft-import) and byte-validate (zk-to-raft-validate). Payloads reuse existing JSON serdes. 4. Cut over: either a short write-fence + re-export/import, or zk-to-raft-live-migrate (re-scan {path,mzxid} diffs until quiet, then fence + final scan). No ZooKeeper watches as CDC — watches coalesce and are not a reliable change feed. 5. Point bootstrap.servers / seed nodes at Raft coordinators; decommission ZK after soak. Live-migrate is a downtime shrink, not a correctness substitute for validation. Prefer stop-the-world if the metadata tree is huge or writes never quiet. Rollback while ZK data is still intact: stop Raft coordinators, restore zookeeper.address, restart. New / default clusters: no ZooKeeper. Format coordinator.raft.dir once, set seed nodes + voters, start coordinators then tablets. ---------------------------------------------------------------------- Performance (JMH, 1-node loopback) ---------------------------------------------------------------------- fluss-jmh compared embedded Curator TestingServer (ZK) vs 1-voter embedded Ratis on the same machine, same flags (2026-08-16 / 08-17). This is not a 3-voter same-AZ NVMe cluster. Production Raft writes still need majority fsync + RTT; the <5ms p50 write target in the design is that production bar, not this loopback run. The paths that dominate coordinator metadata traffic are faster on Raft: Reads (store, in-process KvStore vs ZK RPC) getTable ZK 102 µs Raft 1.6 µs ~64x getDatabase ZK 101 µs Raft 1.0 µs ~100x prefix list / scan ZK ~100 µs Raft 1-3 µs Reads (Admin / RPC, warm cluster) getTableInfo ZK 421 µs Raft 89 µs ~4.7x listTables ZK 283 µs Raft 70 µs ~4x listPartitionInfos ZK 18.5 ms Raft 500 µs ~37x Sequence IDs (Raft reserves a block per append; default coordinator.raft.id-block-size=1000) ZK 1:1 CAS 4871 µs/id (~205 ops/s) Raft blockSize=1 8194 µs/id (one append per id; loses) Raft blockSize=64 127 µs/id (~7.9k ops/s) Raft blockSize=1000 8.5 µs/id (~590x vs ZK) Multi-key / pipelined DDL (appendAll: several keys, one apply wait) Admin createTable ZK 40.3 ms Raft 9.3 ms ~4.3x Admin dropTable ZK 89.3 ms Raft 43.7 ms ~2x Admin createPartition ZK 28.8 ms Raft 17.0 ms ~1.7x store registerFirstSchema ZK 20.4 ms Raft 8.8 ms store registerPartition ZK 14.9 ms Raft 8.8 ms store registerLeaderAndIsr ZK 15.6 ms Raft 8.5 ms Single-key store writes on this 1-voter box are not faster today: Raft still fsyncs the local log, so registerTable is ZK 4.7 ms vs Raft 8.5 ms. A 3-voter quorum adds majority RTT on top of that. That is the physical floor of one durable Raft append (majority fsync + RTT), not a claim that the write path is finished. There is still room to close the gap without weakening commit-apply: - Pipeline / group-commit. JMH is @Threads(1) + append().get(), which empties the Ratis log queue and forces an fsync per record. Concurrent or batched appends can share a flush (force.sync.num / async-flush). We have not turned on unsafe-flush; durability stays on the Raft log fsync. - More appendAll. 2026-08-17 already folded multi-key DDL into one apply wait (createTable 24.7 ms -> 9.3 ms; registerPartition 17.0 ms -> 8.8 ms). Other multi-mutation coordinator paths can do the same. - Topology. The design write target is <5 ms p50 on 3-voter same-AZ NVMe (majority fsync + RTT <1 ms). The 8.5 ms number is 1-node loopback on a laptop disk, not that bar. - Not in v1: leader pre-apply (rejected for correctness; may be revisited post-GA behind a flag) and follower read-index / leader leases. So: reads, lists, block ID allocation, and pipelined DDL are already clearly faster than ZK. A single durable append is bounded by fsync/RTT and still has optimization headroom; it is not the reason to keep ZooKeeper. Leader election is not in the 1-node JMH. RaftLeaderElectionTimingITCase (3-voter loopback) soft-asserts <5s under CI; the design target is <1s with election.timeout=1s, vs historical ZK latch times around ~5s. Full tables and caveats: wiki design §8.4. ---------------------------------------------------------------------- Rejected alternatives (brief) ---------------------------------------------------------------------- - Keep ZooKeeper as the long-term default: does not solve deploy or ops cost; contradicts the published architecture direction. Kept only as a deprecated dual-support backend until Phase 6. - External etcd / Consul: same problem class (a second stateful system). The goal is embedded coordination inside CoordinatorServers. - Use Kafka KRaft as the engine: tightly coupled to Kafka's controller/record stack. We borrow patterns, not the codebase. - Hand-written Raft: high correctness risk when Ratis already provides transport and log. - Every node is a voter (tablets in the Raft group): elastic TabletServer scale would churn quorum membership. Small, stable coordinator quorum; tablets as observers. - GooseFS-style pre-apply (mutate memory before the append commits): rejected for v1. If leadership is lost mid-flight the in-memory store diverges from the log. Pure commit-apply only. - Follower-served strongly consistent reads (read-index / leader leases) in v1: possible later. v1 is leader-only serving plus an eventually consistent metadata stream for caches. ---------------------------------------------------------------------- Implementation status on the branch ---------------------------------------------------------------------- On feature/zookeeper-removal-raft, Phases 1-5 are implemented: dual-support, embedded Ratis + KvStore, metadata stream, TabletServer register/heartbeat/fence, bin/fluss, ZK-to-Raft export/import/validate and live-migrate, Raft TLS/metrics/add-remove-voter/unsafe-recover, optional gossip, preview docs, and Raft as default when neither key is set. Phase 6 (delete ZK mode) is not started. ZK mode remains supported (deprecated). Test coverage already on the branch includes dual-backend HA (DualBackendFlussClusters + @EnumSource(MetadataBackend)), Raft correctness (election timing, stale-leader append reject, minority cannot commit, metadata stream chaos, TabletServer liveness/fence), CLI guards, and migration-tool semantics. Release bar is ./mvnw verify on the affected modules; dual-backend ITs must stay green for both backends until Phase 6. Open (optional, not claimed done): corrupt-snapshot install, disk-full voter, 3-voter production-topology perf. The JMH in the design is 1-node loopback (embedded Curator TestingServer vs single-voter Ratis), not a 3-voter same-AZ NVMe cluster. Production Raft writes still need majority fsync + RTT. ---------------------------------------------------------------------- Points I would especially like feedback on ---------------------------------------------------------------------- 1. Merge/release shape: is accepting Phases 1-5 now, and deferring Phase 6 (delete ZK) to a later major, the right contract? 2. Default backend: neither-key = Raft. Is the upgrade rule ("existing ZK clusters must keep an explicit zookeeper.address") clear enough, or should we require coordinator.seed.nodes to opt into Raft for one more release? 3. Engine choice: Apache Ratis, with KRaft patterns layered on top. Any objection to embedding Ratis vs another library? 4. Voter set: only CoordinatorServers. TabletServers stay observers and never join the Raft group. 5. v1 consistency: leader-only serving; no follower read-index / leader leases in the first release. 6. Migration: stop-the-world export/import as the safe default; live-migrate as a downtime shrink. Is that the ops story we want to document? 7. Gossip: default off; static coordinator.raft.voters remains the source of truth for quorum membership. 8. unsafe-recover: explicit --i-understand-data-loss, offline, majority-loss only. Any additional guardrails before we merge that CLI? 9. Performance: 1-node JMH shows Raft much faster on reads, prefix lists, block ID allocation, and multi-key DDL. A single-key append is slower on loopback because of local fsync + awaitApplied; that path is still open to pipeline / group-commit / same-AZ NVMe measurement, not a v1 pre-apply change. Does that match what we want to tell operators? The full class-level design, sequence diagrams, config appendix, test plan, and rejected alternatives are on the wiki page linked above. Looking forward to your feedback. -- forwardxu
