ZooKeeper

ZooKeeper Basics

Data Model: ZNodes

1. Persistent ZNodes: These nodes exist until explicitly deleted. In our chat app, we would use persistent nodes for configuration data like maximum message size or rate limiting parameters.

   # Store the maximum message size in bytes
   create /chat-app/config/max_message_size "1024"

2. Ephemeral ZNodes: These are automatically deleted when the session that created them ends (whether through client disconnection or timeout). This is perfect for tracking which servers are alive and which users are online.

   # Server 2 registers itself when it starts up
   create -e /chat-app/servers/server2 "192.168.1.102:8080"
   
   # When Bob connects to Server 2, it creates:
   create -e /chat-app/users/bob "server2"
   
   # If Server 2 crashes, both nodes automatically disappear!

3. Sequential ZNodes: These have an automatically appended monotonically increasing counter to their name. In our chat app, we could use this for ordering messages or implementing distributed locks.

   # Alice sends a message to the global chat
   create -s /chat-app/channels/global/msg- "Alice: Hello everyone!"
   # Creates /chat-app/channels/global/msg-0000000001

/chat-app
  /servers           # Directory of available servers
    /server1         # Ephemeral node containing "192.168.1.101:8080" the location of this server
    /server2         # Ephemeral node containing "192.168.1.102:8080" the location of this server
    /server3         # Ephemeral node containing "192.168.1.103:8080" the location of this server
  /users             # Directory of online users
    /alice           # Ephemeral node containing "server1" the server that alice is connected to
    /bob             # Ephemeral node containing "server2" the server that bob is connected to
  /config            # Application configuration
    /max_users       # Persistent node containing "10000" the maximum number of users allowed
    /message_rate    # Persistent node containing "100/sec" the maximum number of messages per second allowed

Server Roles and Ensemble

1. Leader: One server is elected as the leader and is responsible for processing all update requests. When Server 2 registers a new user connection in ZooKeeper, this write request goes to the leader.
2. Followers: The rest of the servers follow the leader's instructions and help serve read requests. When Server 1 needs to find where Bob is connected, it can read this information from any ZooKeeper server in the ensemble.

// Chat Server 1 connecting to ZooKeeper
ZooKeeper zk = new ZooKeeper("zk1:2181,zk2:2181,zk3:2181", 
                             3000, /* session timeout */ 
                             watcher /* callback */);

Watches: Knowing When Things Change

1. When Server 1 starts up, it sets a watch on the /chat-app/users directory:

   // Server 1 watching for user changes
   zk.getChildren("/chat-app/users", true, null);

2. When Bob disconnects from Server 2 and reconnects to Server 3, Server 3 updates Bob's ZNode:

   // Server 3 updates Bob's location
   zk.setData("/chat-app/users/bob", "server3".getBytes(), -1);

3. ZooKeeper automatically notifies Server 1 about this change through its watcher callback:

   // Server 1's watcher callback
   public void process(WatchedEvent event) {
       if (event.getType() == EventType.NodeDataChanged && 
           event.getPath().equals("/chat-app/users/bob")) {
           // Get updated server for Bob
           byte[] data = zk.getData("/chat-app/users/bob", true, null);
           String bobsServer = new String(data);
           // Update routing table: Bob is now on Server 3
           routingTable.put("bob", bobsServer);
       }
   }

Key Capabilities

ZooKeeper for Configuration Management

/chat-app/config/max_message_size "1024"
/chat-app/config/message_rate "100/sec"

1. Update a single ZNode: set /chat-app/config/enable_reactions "true"
2. All chat servers watching this node receive a notification
3. Servers update their behavior without restarting

/ecommerce
  /config
    /pricing_algorithm "dynamic_v2"  # Switch pricing algorithm across all services
    /discount_threshold "50.00"      # Update discount threshold in real-time
    /maintenance_mode "false"        # Toggle maintenance mode off/on

ZooKeeper for Service Discovery

create -e /chat-app/servers/server2 "192.168.1.102:8080"

/streaming
  /services
    /video-transcoder
      /instance1 "10.0.0.1:8080"
      /instance2 "10.0.0.2:8080"
    /recommendation-engine
      /instance1 "10.0.1.1:9000"
      /instance2 "10.0.1.2:9000"
    /payment-processor
      /instance1 "10.0.2.1:5000"

1. Reads /streaming/services/video-transcoder children
2. Connects to one of the available transcoder instances
3. Sets a watch to be notified of any changes to the available transcoders

ZooKeeper for Leader Election

1. Each server creates a sequential ephemeral node under a designated path
2. The server with the lowest sequence number becomes the leader
3. All other servers watch the node with the next lowest sequence number
4. If a leader fails, its node disappears, and the next server in sequence steps up

# Server 1 creates:
create -s -e /chat-app/leader/node- "server1"  # Creates /chat-app/leader/node-0000000001

# Server 2 creates:
create -s -e /chat-app/leader/node- "server2"  # Creates /chat-app/leader/node-0000000002

# Server 3 creates:
create -s -e /chat-app/leader/node- "server3"  # Creates /chat-app/leader/node-0000000003

ZooKeeper for Distributed Locks

1. Each client trying to acquire the lock creates a sequential ephemeral node under a lock path
2. All clients sort the nodes by sequence number
3. The client with the lowest sequence number holds the lock
4. Each client watches the node with the next lower sequence number
5. When a client releases the lock (or crashes), its ZNode is removed, and the next client is notified

# Client 1 wants to send a message:
create -s -e /chat-app/locks/send_message- "client1"  # Creates /chat-app/locks/send_message-0000000001

# Client 2 also wants to send:
create -s -e /chat-app/locks/send_message- "client2"  # Creates /chat-app/locks/send_message-0000000002

# Client 1 sends its message, then deletes its node
delete /chat-app/locks/send_message-0000000001

# Client 2 is notified and now holds the lock

How ZooKeeper Works

Consensus with ZAB

1. Leader Election: When the ZooKeeper ensemble starts or the current leader fails, servers use a voting process to elect a new leader. The primary factor in leader election is having the most up-to-date transaction history. If multiple servers have equivalent transaction histories, then the server with the highest ID will be preferred. (Confusingly, this is the opposite of the application-level leader election pattern we saw earlier, where the node with the lowest sequential ZNode number becomes the leader—a perfect example of how ZooKeeper's internal mechanisms can differ from the patterns built on top of it.)
2. Atomic Broadcast: Once a leader is elected, all write requests go to the leader. The leader then broadcasts these changes to all followers. A write is only considered successful when a majority (quorum) of servers have persisted the change.

1. Server 2 → ZooKeeper Leader: "Create a ZNode at /chat-app/users/bob with value 'server2'"
2. ZooKeeper Leader → All Followers: "Let's add this new ZNode for Bob"
3. Followers → Leader: "Change accepted and saved" (majority needed)
4. Leader → Server 2: "Bob's ZNode has been created successfully"

Strong Consistency Guarantees

1. Sequential Consistency: Updates from a client are applied in the order they were sent. If a client updates node A and then node B, all servers will see the update to A before the update to B.
2. Atomicity: Updates either succeed or fail completely. There are no partial updates.
3. Single System Image: Regardless of which server a client connects to, it will see the same view of the system (after synchronization). A client can read from any server, but all writes go through the leader.
4. Durability: Once an update is applied, it persists until overwritten by a client. Even if servers fail and restart, the updates are not lost.
5. Timeliness: The system guarantees that clients' views of the system are updated within a bounded amount of time.

Read and Write Operations

1. Read Operations: Any server in the ensemble can serve read requests directly from its in-memory copy of the data. This provides high throughput and low latency for reads, which is why ZooKeeper excels at "read-dominant" workloads with ratios of around 10:1 reads to writes.
2. Write Operations: All write requests must go through the leader, which coordinates the update across the ensemble using the ZAB protocol. This centralized approach ensures consistent ordering but makes writes more expensive than reads.

• When Server 1 needs to find where Bob is connected (a read operation), it can query any ZooKeeper server.
• When Server 3 needs to update Bob's location (a write operation), this request goes to the leader, which then synchronizes it across the ensemble.

Sessions and Connection Management

1. Session Establishment: When a client connects to ZooKeeper, it establishes a session with a configurable timeout (typically 10-30 seconds).
2. Heartbeats: The client sends periodic heartbeats to maintain its session. If ZooKeeper doesn't receive a heartbeat within the timeout period, it assumes the client has failed.
3. Session Recovery: If a client loses its connection to a ZooKeeper server, it can connect to a different server and recover its session, as long as it does so before the session times out.
4. Session Expiration: If a session expires, all ephemeral nodes created by that client are automatically deleted, and all watches registered by that client are removed.

1. Server 2 establishes a session with ZooKeeper when it starts
2. It creates ephemeral nodes for itself and connected users
3. If Server 2 crashes, its session eventually times out
4. ZooKeeper automatically removes all ephemeral nodes owned by Server 2
5. Other servers watching these nodes are notified about users going offline

Storage Architecture

1. Transaction Log: Every state change (transaction) is first written to a transaction log on persistent storage. This write-ahead logging ensures that no acknowledged update is ever lost, even if a server crashes immediately after confirming the update.
2. Snapshots: Periodically, ZooKeeper creates snapshots of its in-memory database to speed up recovery. When a server restarts, it loads the most recent snapshot and then replays transaction logs to recover the complete state.

"The most performance-critical part of ZooKeeper is the transaction log. ZooKeeper must sync transactions to media before it returns a response. A dedicated transaction log device is key to consistent good performance."

Handling Failures

1. Server Failures: If a follower fails, the leader continues to operate as long as a quorum of servers remains available. If the leader fails, a new leader election occurs automatically by promoting the follower with the highest ID.
2. Network Partitions: If the network is partitioned such that no group has a majority of servers, ZooKeeper will not process write requests until the partition is resolved. This prevents the "split-brain" problem where different parts of the system disagree about the current state.
3. Client Failures: If a client fails, any ephemeral nodes it created are automatically removed after its session times out. This would be important for our chat application - if Server 2 crashes, all users connected to it are automatically marked as offline when their ZNodes disappear.
4. Client Session Management: ZooKeeper tracks client sessions and provides mechanisms for clients to recover and reconnect after temporary disconnections, without losing their session state or watches.

1. Server 2 crashes
2. Server 2's ZooKeeper session times out (typically after 10-30 seconds)
3. All ephemeral nodes created by Server 2 are automatically deleted, including:
   - /chat-app/servers/server2
   - All /chat-app/users/X where X was connected to Server 2
4. Other servers with watches on these nodes receive notifications
5. They update their routing tables to mark users as offline
6. When a new Server 3 comes back online, it creates a new session and registers itself

ZooKeeper in the Modern World

Current Usage in Major Distributed Systems

Alternatives to Consider

Limitations

So when should you use ZooKeeper then?

Smart Routing

Certain Infrastructure Design Problems

Durable Distributed Locks

Summary

References

• Apache ZooKeeper Getting Started Guide
• Apache ZooKeeper Internals Documentation
• ZooKeeper: Wait-free coordination for Internet-scale systems