The Proposal: A Decentralized Architecture For Metadata Management

[Migorithm]

Maintaining separate metadata store will introduce:

• complexity
• maintainability issue

And this is not only when we develop the project but also for those who would potentially tab into the project.

Assigning a dedicated controller role to some node(or group of) can be an option, and in fact that's a strategy taken by Kafka 4.0

That has two issues:

• Scalability issue by controller being a bottleneck
• Operability issue as external service is required to move existing data

So, In this discussion, I suggest the followings:

• Scalable Weakly-consistent Infection-style Membership(SWIM)
• Dynamically-Sharded Replicated State Machine(DS-RSM)

SWIM

SWIM was created to solve the scaling limitations of traditional "heartbeat" protocols (where every node checks on every other node, which causes network congestion as the cluster grows).

SWIM has two main components, and this is where the distinction lies:

1. Failure Detection (The "Probe")
   This part is unique to SWIM and is not standard gossip.

• Direct Ping : Node A picks a random Node B and pings it.
• Indirect Ping: If Node B doesn't reply, Node A asks specific other nodes (C and D) to ping B on its behalf (this is called ping-req).
  • you can configure maximum number of peers for dissemination(for example, k=3)

This mechanism prevents false positives (marking a node dead just because the link between A and B is slow).

2. Dissemination (The standard "Gossip")
   Once Node A confirms that Node B is dead (via the Failure Detection phase), it needs to tell everyone else.

Instead of broadcasting to the whole cluster "B IS DEAD!", Node A piggybacks this information onto its future messages to other nodes(e.g., when it probes).

This information spreads "infection-style" (gossip) through the cluster until everyone updates their membership list.

• Scenario: Node A knows that Node C is dead.
• Action: When Node A sends its regular Ping to Node B, it adds a tiny footer to the message: "By the way, C is dead."
• Cost: This adds a few bytes to a packet that was going to be sent anyway. It does not create a new packet.

In essence, SWIM is to maintain the membership list in the cluster.
We can consider handling the following event types:

• Alive(Join/Heartbeat)
  • Trigger: When a node boots up/refutes a suspicion
  • Payload: Node ID, Node's physical attributes (IP, Port, Rack), Incarnation Number(more on in the next subsection).
• Suspect(Failure detection)
  • Trigger: A missed ping - this puts the node in probation
• Confirm(Dead)
  • Trigger: timeout after Suspect state. All nodes remove the node in question from their ring calculations.

And as "Piggybacking", we can introduce the following event types as well:

• ShardLeader (Metadata Routing)
  • Trigger: When a Raft election completes within a DS-RSM group.
  • Payload: ShardID, Leader Node ID, TermID.
  • Purpose: Tells the cluster (and proxies) exactly which node to contact for metadata regarding a specific hash range. The TermID is crucial here to ignore old claims from deposed leaders.

Preventing "Zombie Loop" Using Incarnation Number

Without Incarnation Numbers, distributed failure detection is impossible to stabilize.

• Node B can’t reach Node A. Node B gossips: "Node A is Suspect!"
• Node A hears this. It shouts back: "No! I am Alive!"
• Node C hears both messages. Which one is true?
• If C trusts "Alive", it might try to connect to a dead node.
• If C trusts "Suspect", Node A keeps getting kicked out of the cluster even though it is healthy.

The Incarnation Number acts as a timestamp. The cluster follows a strict logic hierarchy when comparing two messages about the same node.

It follows the simple rule:

• Higher Incarnation wins. (e.g., Inc: 5 beats Inc: 4).
• If Incarnations are equal, specific states override others:
  • Confirm (Dead) overrides Suspect.
  • Suspect overrides Alive.

For more on SWIM, read this paper.

Dynamically-Sharded Replicated State Machine(DS-RSM)

Rather than maintaining a monolithic metadata store (like a global Etcd/Zookeeper) or a single heavy Controller Quorum (like Kafka KRaft), we distribute the state management responsibility across the nodes themselves.

How it achieves that is as follows:

• Dynamic Sharding: Metadata is partitioned (sharded) based on resource keys (e.g., Topic ID). These shards are distributed among the compute nodes.

• Localized Consensus: Instead of one global Raft group reaching consensus for every operation in the system, it forms small, dynamic Raft groups (state machines) only for the nodes involved in a specific shard.

  • Example: If Node A, B, and C host "Shard 1", only they participate in the consensus for "Shard 1". Node D doesn't care.

• Elimination of External Dependencies: This removes the need for an external KV store. The application nodes are the data store for their own metadata.

• Linear Scalability: As you add nodes, you automatically add more CPU/Disk capacity for metadata processing. The "Controller" capacity scales linearly with the "Data" capacity.

How it interplays

Decision

• Metadata Shard #45 (The Raft Group) decides: "Topic-Blue is assigned to Node-Y."
• This is written to the Raft Log and replicated to the 3 nodes in Shard #45.
• This is the only place in the universe where this fact is guaranteed to be true.

Propagation

The worker node (say, Node-Y) needs to know it has work to do. It does not "poll" the shard.

It's the Shard Leader that dictates it.

• Action: The Leader of Shard #45 sends a direct RPC command to Node-Y.
  • Command: OpenTopic(Topic-Blue, Epoch=1)
• Result: Node-Y spins up the resources for that topic and replies OK.
• State: Node-Y now has a local in-memory flag: "I own Topic-Blue."

Propagation Clients (The "Pull + Cache")

This is where the scaling magic can happen.
We do not broadcast topic assignments to all brokers. (Broadcasting 1 million topic assignments would kill the network).

Instead, it can be implemented as "need-to-know" basis(lazy loading)

• Client asks Node-B: "Connect me to Topic-Blue."
• Node-B checks its Cache: Empty.
• Node-B calculates hash (Hash(Topic) = 123). It consults its local SWIM-maintained Ring Map to see that hash 123 belongs to Shard #45 (Leader IP: 10.0.0.5).
• Shard #45 replies: "Node-Y."
• Node-B caches this: Topic-Blue -> Node-Y.
• Node-B routes the data to Node-Y.

Handling Stale Cache

This is the most important part. What happens if Shard #45 changes its mind and moves the topic to Node-Z,
but other nodes still think it is on Node-Y?

This is called a stale cache, and it is resolved via error handling:

• The Move: Shard #45 moves the topic to Node-Z. Node-Y is told to close it (or crashes).

• The Mistake: Node-B (using stale cache) tries to send data to Node-Y.

• The Rejection:

  • Scenario A (Node-Y is alive): Node-Y receives the data. It checks its local list. "I don't own Topic-Blue anymore." It returns a specific error: NotLeaderForTopic.
  • Scenario B (Node-Y is dead): The connection times out.

• The Fix:

  • Node-B receives the error.
  • Node-B invalidates its cache entry for Topic-Blue.
  • Node-B asks Shard #45 again: "Who owns Topic-Blue?"
  • Shard #45 replies: "Node-Z."
  • Node-B updates cache and retries.

[seonWKim]

Cassandr uses a peer-to-peer architecture with no dedicated controller nodes. We may borrow some ideas from them.

• Gossip protocol with phi accrual failure detection
• No Raft groups (all replicas have the same role). The coordinator (the node that first receives the request) sends writees to all replicas. Uses quorums.

[Migorithm]

Cassandra recently moved away from Gossip and adopted Transaction Cluster Metadata which is again, backed by Raft. The only problem is it assumes only one subgroup (by assigning role to a node, say "Cluster metadata service(CMS)").

This funnels all the metadata changes through a single Raft group(CMS), and this introduces a bottleneck - which is essentially the same problem as Kafka 4.0

Yes, metadata may be low-velocity so it wouldn't matter up to a certain point.

We are aiming at building scalable system though.

[seonWKim]

Wonder why they've moved to centralizing their metadata(maybe even big systems don't need that scalability) 🤔

I've been thinking about your "Dynamically-Sharded Replicated State Machine(DS-RSM)" suggestion, and couldn't find a solution for the following case:

• When a new node is added/removed how should we add/remove it to/from a Raft group
• Who should decide on how to add/remove from a Raft group? If not, what algorithm?

[Migorithm]

In DS-RSM model, we can avoid a global controller by making every Shard Leader a "Local Controller" for its own tiny universe. So basically, this is "Zero-controller" approach.

Since we don't have a central brain to assign nodes, you replace the "Brain" with a "Function". Everyone must agree on the algorithm (e.g., Consistent Hashing or Rendezvous Hashing).

To your question,

When a new node is added/removed how should we add/remove it to/from a Raft group

Let's break down DS and RSM from "DS-RSM".

Dynamic Sharding(DS)

Instead of storing all cluster metadata in one place, the metadata is broken down into smaller chunks called vNodes (Virtual Nodes).

• Consistent Hashing: These vNodes are distributed across the available physical servers using consistent hashing.
• Distribution: If you have metadata for 100,000 topics, DS-RSM splits them so that Server A manages topics 1–10,000, Server B manages 10,001–20,000, and so on.
• No Single Leader: There is no single "cluster leader" that manages everything. Every server acts as the metadata leader for a specific slice of the cluster.

Replicated State Machine(RSM)

Each vnode operates as an independent Replicated State Machine backed by a consensus algorithm (typically Raft).

• Fault Tolerance: Every vNode has replicas. If the server managing "vNode 1" dies, one of its followers on another server immediately steps up as the leader for that specific shard of metadata.

• Strong Consistency: Because it uses Raft, any change to metadata (e.g., "Topic A is now sealed") is guaranteed to be consistent and durable before it is acknowledged.

How to determine the follower

Selection of followers is not a random choice made by a leader; Rather, this is deterministic calculation based on the consistent hash ring topology.

Here is specifically how the system decides which physical servers will act as the followers (replicas) for a specific vNode using the ring.

1. Clockwise Walk
   The standard algorithm for selecting followers in a consistent hash ring works by "walking" the ring. To find the followers, the system continues walking clockwise along the ring from the Leader's position. The next distinct servers that hold the valid tokne become the followers - which is called Clockwise Walk

For a replication factor of 3 (1 Leader + 2 Followers), the system picks the first 3 distinct servers following the vNode's hash on the ring.

2. Constraint Filtering (Label Awareness)
   In production systems, a simple clockwise walk is dangerous. If Server A, Server B, and Server C are physically next to each other in the same server rack, a single power outage could wipe out all your replicas. To prevent this, the selection algorithm adds a constraint filter to the ring walk:

• Pick Node 1 (Leader): Found at position 100. (Location=Rack 1).
• Walk to Node 2: Found at position 120. (Location=Rack 1).
  • Decision: SKIP.
• Walk to Node 3: Found at position 150. (Location: Rack 2).
  • Decision: SELECT as Follower 1.
• Walk to Node 4: Found at position 180. (Location=Rack 3).
  • Decision: SELECT as Follower 2.

---

Given all of these, we can picture Raft groups in DS-RSM as:

A node is always a leader of certain shard

But at the same time, the same node is a replica of some other shard.

To put it in another way, Server A might be:

• The Leader for Shard #1 (handling writes for Topic A).
• A Follower for Shard #2 (replicating data for Topic B).
• A Follower for Shard #3 (replicating data for Topic C).

[seonWKim]

Simple notes before going to work.

• Client request flow
  • Client can send request to any node (let’s say it’s node A)
  • Node A:
    • If Node A is the shard leader: handles the request
    • If Node A is not the shard leader: pass the request to Node B which can handle the request
      • Node A holds shard mappings, for example Shard 2 is mapped to Node B
  • Node B:
    • If Node B is the shard leader: handles the request
    • If Node B is not the shard leader
      • This may happen when Node A has stale metadata. We may notify Node A that it has stale metadata
      • Pass the request to Node C which can handle the request(recursive)
• Node addition
  • When node is added, we add vNodes into the consistent hash ring => 2 cases. When the newly added node becomes a leader / follower
  • Affected Raft Groups
    • Should we freeze the raft groups for rebalancing?
    • New vNode
      • When it’s selected as the follower: receive replicated data
      • When it’s selected the leader: Can a new vNode be a leader?
• Node deletion
  • When node is added, we delete vNodes from the consistent hash ring
  • Affected Raft Groups
    • vNode who becomes the leader: simply change the role would be enough?
    • vNode who becomes the follower: can a leader become follower in this case?
• What metadata should each node hold?
  • vNode -> segment
  • which node holds the token

[seonWKim]

I found that we have to fix the update_member logic. Our implementation currently makes state with higher incarnation number as the winner, but the paper states that Confirm state(for our implementation, it's Dead state) should override other states regardless of the incarnation number.

This is due to the fact that Dead state is only emitted after a node was suspected and failed to refute it within the suspicion timeout. The node already had a change to send Alive but it didn't, so the protocol can treat this as definitive failure.
Also this logic protects against out-of-order messages. In our weakly consistent system, old messages can arrive late. If Alive could override Dead, a delayed Alive could resurrect a dead node. Making Dead absolute guarantees failure information propagates for a long time.

[seonWKim]

Example case of what could happen when we make Dead node resurrect:

• Node A suspects node B and starts the suspicion timer.
• B is actually partitioned and cannot respond.
• The suspicion timeout expires → A marks B as Dead and gossips this to the cluster.
• Meanwhile, B had previously sent an Alive(inc=5) message that was delayed in the network. This delayed Alive(inc=5) arrives after some nodes have already accepted Dead.

If our update logic prefers higher incarnation over state, those nodes would revert B from Dead → Alive, resurrecting a node that the cluster already decided to remove.

By making Dead absolute:

• Once any node accepts Dead, that decision cannot be undone
• Dead gossip continues to spread
• Eventually, all nodes converge on removing B

This guarantees that failure information propagates permanently, even in the presence of message reordering and delays.