Apache Pulsar & BookKeeper: Tiered Storage and Replicated Log Architecture

1. Context & Architecture Reality

Project Identity, Version & License Baseline

SYSTEM ARCHITECTURE BLUEPRINTDISTRIBUTED SYSTEMS

⤢Expand

ARCHITECTURAL FLOW · STEP 1 OF 3

1. LAYER 2: STATEFUL LOG STORAGE (BookKeeper Bookies) & LAYER 3: TIERED STORAGE OFFLOAD (Amazon S3 / GCS)

Establishes the ingress and structural boundary across LAYER 2: STATEFUL LOG STORAGE (BookKeeper Bookies) & LAYER 3: TIERED STORAGE OFFLOAD (Amazon S3 / GCS): Append-only log ledgers. Bookies store chunks independently of topic ownership. Zero partition rebalancing copies when adding new storage nodes!; Automated async offload of sealed ledgers: Infinite retention at 90% lower cost; Dedicated high-speed NVMe.

Architectural Invariant: Sub-millisecond p99 latency Purged once EntryLog is synced.

ARCHITECTURAL FLOW · STEP 2 OF 3

2. Stateless Serving vs Distributed Storage & Segment-Centric Ledgers vs Monolithic Partitions

Coordinates core data-plane execution and state transitions through Stateless Serving vs Distributed Storage & Segment-Centric Ledgers vs Monolithic Partitions: Ledger L1 (Entries 0..999) → Bookies [B1, B2, B3] Ledger L2 (Entries 1000..1999)→ Bookies [B2, B3, B4] Ledger L3 (Entries 2000..2999)→ Bookies [B3, B4, B5]; Rebalance is instantaneous: Update ledger pointers in metadata in < 5ms!.

Architectural Invariant: Rebalance is instantaneous: Update ledger pointers in metadata in < 5ms!.

ARCHITECTURAL FLOW · STEP 3 OF 3

3. BookKeeper Quorum Math: E, W, and A & Quorum Parameters: E=3 (Ensemble), W=3 (Write), A=2 (Ack)

Enforces durability, state convergence, and output verification across BookKeeper Quorum Math: E, W, and A & Quorum Parameters: E=3 (Ensemble), W=3 (Write), A=2 (Ack): Stripes entries across Bookies with LastAddConfirmed (LAC) consistency. Ensemble Size (E): Total pool of bookies assigned to the active ledger. Write Quorum (W): Number of bookies each individual entry is replicated to. Ack Quorum (A): Number of acknowledgments required before client return; LastAddConfirmed (LAC): Piggybacked on writes; readers only read entries <= LAC.

Architectural Invariant: Stripes entries across Bookies with LastAddConfirmed (LAC) consistency. Ensemble Size (E): Total pool of bookies assigned to the active ledger. Write Quorum (W): Number of bookies each individual entry is replicated to. Ack Quorum (A): Number of acknowledgments required before client return.

In modern enterprise data platforms, distributed event streaming is the nervous system of the enterprise. For over a decade, Apache Kafka served as the industry standard. Kafka introduced the concept of the partitioned append-only commit log, replacing fragile point-to-point queues with high-throughput streaming pipelines.

However, as enterprise datasets expanded into petabytes and cloud-native Kubernetes environments became standard, Kafka’s tightly coupled architecture exposed severe operational pain points:

• Coupled Compute & Storage: In Kafka, each broker acts as both the serving layer (handling client TCP connections and partition assignment) and the storage layer (storing log segments on local physical disks).
• The Partition Rebalance Nightmare: If a partition grows to $5\text{ TB}$ and a broker dies or disk fills up, rebalancing that partition onto a new broker requires copying $5\text{ TB}$ of data across the network, saturating cluster bandwidth and risking cascading cluster-wide slowdowns.
• Expensive Long-Term Retention: Storing months of historical data for auditing or machine learning requires provisioning massive, expensive NVMe SSD storage pools directly on compute brokers.

In 2012, Yahoo! engineers designed and open-sourced Apache Pulsar and Apache BookKeeper to solve these exact architectural limitations. Pulsar completely divorces stateless serving from stateful storage, creating a resilient, cloud-native streaming engine.

+--------------------------------------------------------------------------------------------------+
|                            PULSAR & BOOKKEEPER DECOUPLED ARCHITECTURE                            |
|                                                                                                  |
|   Producers / Publishing Clients                                                                 |
|        |                                                                                         |
|        v                                                                                         |
|   +-----------------------------------------------------------------------------------------+    |
|   | LAYER 1: STATELESS BROKERS (Apache Pulsar)                                              |    |
|   | - Client TCP termination, Authentication (JWT), Topic routing, In-memory message cache  |    |
|   | - ZERO persistent disk state! Any broker can assume ownership of any topic in < 5ms.    |    |
|   +-----------------------------------------------------------------------------------------+    |
|        |                                                                                         |
|        v Distributed Ledger Client (Stripes entries across bookies)                              |
|   +-----------------------------------------------------------------------------------------+    |
|   | LAYER 2: DISTRIBUTED LOG STORAGE (Apache BookKeeper)                                    |    |
|   |                                                                                         |    |
|   |   [Bookie 1]         [Bookie 2]         [Bookie 3]         [Bookie 4]         [Bookie 5]    |    |
|   |   +-----------+      +-----------+      +-----------+      +-----------+      +-----------+ |    |
|   |   | Journal   |      | Journal   |      | Journal   |      | Journal   |      | Journal   | |    |
|   |   | EntryLog  |      | EntryLog  |      | EntryLog  |      | EntryLog  |      | EntryLog  | |    |
|   |   +-----------+      +-----------+      +-----------+      +-----------+      +-----------+ |    |
|   +-----------------------------------------------------------------------------------------+    |
|        |                                                                                         |
|        v Automated Background Offloader (Sealed Ledgers)                                         |
|   +-----------------------------------------------------------------------------------------+    |
|   | LAYER 3: TIERED CLOUD STORAGE (Amazon S3 / Google Cloud Storage / Azure Blob)           |    |
|   | Infinite historical retention at 90% lower storage cost; transparently queryable!       |    |
|   +-----------------------------------------------------------------------------------------+    |
+--------------------------------------------------------------------------------------------------+

2. Core Problem & Failure Modes

Designing a multi-tier, decoupled distributed messaging platform introduces subtle systems challenges:

1. Broker Failover & Split-Brain Writes: Because brokers are stateless, when Broker 1 fails (or experiences a temporary network partition), Broker 2 immediately assumes ownership of Topic $T$. If Broker 1 recovers and still believes it is the valid owner, it might attempt to append writes to the underlying storage log. BookKeeper must enforce Ledger Fencing to permanently seal the active ledger and reject stale writes.

2. The Slow Bookie Tail-Latency Problem: If an ensemble requires acknowledgments from multiple storage nodes ($A=2$ out of $W=3$), a single degraded bookie experiencing disk write stalls can drag down cluster write throughput. BookKeeper mitigates this using Speculative Write Re-routing: if a bookie does not acknowledge a write within a configured threshold (e.g., $10\text{ ms}$), the client automatically dispatches a speculative write to an alternate bookie in the ensemble.

3. EntryLog Garbage Collection Churn: BookKeeper multiplexes entries from thousands of independent ledgers into shared sequential EntryLog files. When some ledgers expire or are deleted by Pulsar retention policies, the EntryLog becomes fragmented with dead space. The Bookie must run background compaction threads to rewrite active entries into clean files, introducing background disk read/write amplification.

4. Cursor Backlog Memory Bloat: In Pulsar, every consumer subscription maintains an independent cursor tracking its acknowledged message offsets. If an offline batch consumer stops reading for 5 days, messages cannot be purged. If millions of backlogged messages are retained on Bookie SSDs, storage costs multiply rapidly unless tiered storage offload is properly configured.

3. Technical Deep Dive

BookKeeper Quorum Striping Mechanics ($E,W,A$)

A topic partition in Pulsar is an abstraction composed of a sequence of Ledgers. A Ledger is an append-only, write-once sequence of entries.

When a Pulsar broker opens a ledger, it configures three quorum parameters:

• Ensemble Size ($E$): The number of distinct Bookies allocated to hold data for this ledger.
• Write Quorum ($W$): The number of Bookies each individual entry is replicated to ($W\le E$).
• Ack Quorum ($A$): The number of Bookies that must acknowledge the write before the broker confirms the write to the producer ($A\le W$).

Round-Robin Striping Formula

Entries are striped across the ensemble. For entry $i$, the write set starts at index $i\mathrm{mod}E$ and spans $W$ consecutive bookies modulo $E$:

Example: E = 4, W = 3, A = 2 across Bookies [B0, B1, B2, B3]

Entry 0:  [ B0,  B1,  B2 ]  --> Requires 2 ACKs
Entry 1:  [ B1,  B2,  B3 ]  --> Requires 2 ACKs
Entry 2:  [ B2,  B3,  B0 ]  --> Requires 2 ACKs
Entry 3:  [ B3,  B0,  B1 ]  --> Requires 2 ACKs

This striping guarantees that:

1. Write I/O is balanced uniformly across all $E$ physical bookie disks.
2. Even if $W-A$ bookies fail or experience high latency, writes proceed without interruption.

LastAddConfirmed (LAC) & Consistency Guarantee

BookKeeper guarantees that readers never read uncommitted or partial entries using the LastAddConfirmed (LAC) protocol.

• When a producer writes Entry $K$, the broker sends Entry $K$ to the bookies. Piggybacked inside Entry $K$‘s header is the current LAC (the highest entry ID that has already achieved Ack Quorum $A$).
• Readers are strictly prohibited from reading any entry beyond the current LAC:

Because LAC updates are piggybacked on ordinary write RPCs, BookKeeper achieves strong consistency without requiring separate coordinator consensus rounds!

Bookie Internal Disk Architecture: Journal vs EntryLog

A Bookie achieves sub-millisecond p99 write latency while supporting heavy historical data reads by physically separating write and read paths:

[Write Request: Entry K]
         |
         +---------------------------------------+
         |                                       |
         v (Sequential fsync)                   v (In-Memory Buffer)
+------------------------+              +------------------------+
| 1. DEDICATED JOURNAL   |              | 2. WRITE CACHE (RAM)   |
| - Fast NVMe SSD        |              | - Aggregates entries   |
| - Append-only log file |              | - Serves tailing reads |
| - Instant Write ACK!   |              +------------------------+
+------------------------+                           |
                                                     v (Periodic Batched Flush)
                                        +--------------------------------+
                                        | 3. LEDGER STORAGE DISKS        |
                                        | - EntryLog: Sequential chunks  |
                                        | - Index: RocksDB ledger offset |
                                        +--------------------------------+

1. Journal (WAL): When a write arrives, it is appended to an open Journal file and immediately fsync’d to disk. Once the Journal acknowledges, the bookie sends its ACK to the broker.
2. Write Cache: Concurrently, the entry is placed in an in-memory sorted write cache. Tailing consumers (reading live messages within seconds) read directly from this cache, never touching disk!
3. EntryLog & Index: Periodically, the write cache flushes large sequential blocks into EntryLog files on separate storage disks, updating a local RocksDB index mapping (ledgerId, entryId) -> (entryLogFile, byteOffset).

4. System Design Patterns & Trade-offs

+----------------------------------------------------------------------------------------------------+
| TRADEOFF MATRIX: PULSAR VS KAFKA ARCHITECTURAL CHOICES                                             |
+--------------------------+------------------------------------+------------------------------------+
| Dimension                | Apache Pulsar                      | Apache Kafka                       |
+--------------------------+------------------------------------+------------------------------------+
| Architecture             | Decoupled (Broker + BookKeeper)    | Monolithic (Broker stores data)    |
| Storage Model            | Segment-Centric (Ledger slices)    | Partition-Centric (Contiguous log) |
| Partition Rebalance Cost | Zero (Metadata pointer update)     | High (Copies gigabytes over network|
| Storage Tiering          | Built-in native S3/GCS offload     | Dependent on proprietary plugins   |
| Operational Complexity   | High (Manages Brokers, Bookies, ZK)| Low/Medium (KRaft mode: no ZK)     |
| Raw Hardware Ingestion   | ~1.2M msgs/sec per cluster         | ~2.0M msgs/sec per cluster (zero-cp|
+--------------------------+------------------------------------+------------------------------------+

5. Practical Implementation

Below is a complete TypeScript simulation demonstrating BookKeeper’s core storage primitives:

1. Quorum write striping across an ensemble ($E,W,A$).
2. LastAddConfirmed (LAC) tracking.
3. Ledger Fencing state machine preventing split-brain writes during broker failover.

export interface LedgerEntry {
  ledgerId: number;
  entryId: number;
  data: string;
  lac: number;
}

export interface BookieStorage {
  bookieId: string;
  entries: Map<string, LedgerEntry>;
  fencedLedgers: Set<number>;
}

/**
 * Apache BookKeeper Cluster Simulator
 */
export class BookKeeperCluster {
  private bookies: BookieStorage[] = [];

  constructor(bookieCount: number = 5) {
    for (let i = 0; i < bookieCount; i++) {
      this.bookies.push({
        bookieId: `bookie-${i}`,
        entries: new Map(),
        fencedLedgers: new Set()
      });
    }
  }

  /**
   * Fences a ledger on all bookies during broker takeover
   */
  public fenceLedger(ledgerId: number): boolean {
    for (const bookie of this.bookies) {
      bookie.fencedLedgers.add(ledgerId);
    }
    return true;
  }

  /**
   * Appends an entry using Quorum Striping (E, W, A)
   */
  public writeEntry(
    ledgerId: number,
    entryId: number,
    data: string,
    currentLac: number = -1,
    ensembleSize: number = 4,
    writeQuorum: number = 3,
    ackQuorum: number = 2
  ): { success: boolean; acks: number; lac: number } {
    const entry: LedgerEntry = {
      ledgerId,
      entryId,
      data,
      lac: currentLac
    };

    // Calculate target bookies for this entry using striping formula:
    // startIdx = entryId % ensembleSize
    const targetBookieIndices: number[] = [];
    const startIdx = entryId % ensembleSize;
    for (let i = 0; i < writeQuorum; i++) {
      targetBookieIndices.push((startIdx + i) % ensembleSize);
    }

    let acks = 0;
    for (const idx of targetBookieIndices) {
      const bookie = this.bookies[idx];

      // Check if ledger was fenced by a competing broker
      if (bookie.fencedLedgers.has(ledgerId)) {
        throw new Error(`Write failed: Ledger ${ledgerId} is FENCED on ${bookie.bookieId}`);
      }

      const key = `${ledgerId}:${entryId}`;
      bookie.entries.set(key, entry);
      acks++;
    }

    const success = acks >= ackQuorum;
    return {
      success,
      acks,
      lac: success ? entryId : currentLac
    };
  }

  /**
   * Reads entries strictly up to LastAddConfirmed (LAC)
   */
  public readEntries(
    ledgerId: number,
    fromEntryId: number,
    lac: number,
    ensembleSize: number = 4,
    writeQuorum: number = 3
  ): LedgerEntry[] {
    const results: LedgerEntry[] = [];

    for (let id = fromEntryId; id <= lac; id++) {
      const startIdx = id % ensembleSize;
      let foundEntry: LedgerEntry | null = null;

      for (let i = 0; i < writeQuorum; i++) {
        const bookieIdx = (startIdx + i) % ensembleSize;
        const entry = this.bookies[bookieIdx].entries.get(`${ledgerId}:${id}`);
        if (entry) {
          foundEntry = entry;
          break;
        }
      }

      if (foundEntry) {
        results.push(foundEntry);
      }
    }

    return results;
  }
}

function runDemo(): void {
  console.log("=== Apache Pulsar & BookKeeper Cluster Architecture ===");
  const cluster = new BookKeeperCluster(5);
  console.log("Initialized Apache BookKeeper ensemble with 5 bookies");
  const w1 = cluster.writeEntry(101, 1, "ledger entry payload #1", -1);
  const w2 = cluster.writeEntry(101, 2, "ledger entry payload #2", w1.lac);
  console.log(`Appended ledger entries 1 and 2 to ensemble: success=${w1.success && w2.success}, currentLAC=${w2.lac}`);
  const entries = cluster.readEntries(101, 1, w2.lac);
  console.log(`Read ${entries.length} verified entries from bookie cluster (up to LAC=${w2.lac})`);
  cluster.fenceLedger(101);
  try {
    cluster.writeEntry(101, 3, "stale zombie broker write", w2.lac);
  } catch (err: any) {
    console.log(`Ledger fencing verified: ${err.message}`);
  }
}

runDemo();

Simulation Execution Output & State Transitions

=== Apache Pulsar & BookKeeper Cluster Architecture ===
Initialized Apache BookKeeper ensemble with 5 bookies
Appended ledger entries 1 and 2 to ensemble: success=true, currentLAC=2
Read 2 verified entries from bookie cluster (up to LAC=2)
Ledger fencing verified: Write failed: Ledger 101 is FENCED on bookie-3

6. Failure Scenarios & Edge Cases

1. The Fenced Broker Zombie Race

Broker 1 experiences a 15-second Linux kernel page-cache freeze. The Pulsar cluster manager reassigns Topic $A$ to Broker 2. Broker 2 opens Ledger 2 and issues FenceLedger(Ledger 1) to the Bookie cluster. When Broker 1 wakes up, its internal queue still contains 50 unwritten customer messages.

• Handling: Broker 1 attempts to append Entry 95 to Ledger 1. The bookies inspect their internal fencedLedgers bitset and reject the write with a FencedException. Broker 1 discovers it was superseded, terminates its producer connections, and yields gracefully.

2. Slow Bookie Speculative Read/Write Triggers

If Bookie 3 encounters an NVMe background wear-leveling cycle, write latency jumps from $1\text{ ms}$ to $400\text{ ms}$. If the broker waits synchronously for Bookie 3, producer latency spikes.

• Handling: BookKeeper clients configure speculativeRequestTimeoutPolicy. If Bookie 3 does not acknowledge within $8\text{ ms}$, the client immediately fires a speculative write to Bookie 4. Whichever bookie responds first satisfies the Ack Quorum, keeping p99 latency flat.

7. Production Operations & Observability

BookKeeper Auto-Recovery (AutoRecoveryMain)

BookKeeper clusters run a dedicated background daemon called Auto-Recovery:

1. An Auditor node detects bookie failures via ZooKeeper/etcd ephemeral session loss.
2. The Auditor identifies all ledgers that had active replicas on the dead bookie.
3. Replication Worker threads read the surviving replicas of those ledgers and replicate missing entries to healthy bookies, automatically restoring quorum health without human intervention.

8. Security & Compliance

1. End-to-End Payload Encryption: Pulsar producers can encrypt message bodies using public-key cryptography before transmission. The message remains encrypted in-flight, inside broker memory, inside BookKeeper disks, and in S3 cold storage. Only consumer applications possessing the private RSA key can decrypt payloads.
2. Multi-Tenant Isolation: Pulsar organizes data into tenant / namespace / topic. System administrators enforce storage quotas, bandwidth rate limits, and encryption keys independently per tenant.

9. Real-World Case Studies

Yahoo! Mail & Financial Platforms

Yahoo! previously ran hundreds of isolated Kafka and RabbitMQ clusters across datacenters. Managing cluster migrations and disk capacity was an operational nightmare.

• Migration to Pulsar: By migrating to a unified multi-tenant Apache Pulsar cluster across four global datacenters, Yahoo! eliminated manual partition rebalancing.
• Tiered Storage Savings: Using Pulsar’s native S3 offloader, Yahoo! reduced message storage hardware expenses by $70\%$, retaining multi-year audit logs seamlessly.

10. Technology Comparison Matrix

+----------------------------------------------------------------------------------------------------+
| DISTRIBUTED MESSAGING SYSTEMS MATRIX                                                               |
+----------------------+--------------------+--------------------+-----------------------------------+
| System               | Architecture       | Storage Engine     | Best Fit                          |
+----------------------+--------------------+--------------------+-----------------------------------+
| Apache Pulsar        | Decoupled 2-Layer  | BookKeeper Ledgers | Multi-tenant cloud-native apps,   |
|                      | (Brokers + Bookies)| + S3 Tiered Offload| zero-rebalance scaling, long reten|
+----------------------+--------------------+--------------------+-----------------------------------+
| Apache Kafka         | Monolithic Broker  | Local OS PageCache | Ultra-high throughput streaming   |
|                      | & Disk Storage     | & Log Segments     | with short/medium retention (<7d) |
+----------------------+--------------------+--------------------+-----------------------------------+
| RabbitMQ             | Erlang AMQP Broker | Mnesia + Disk Queue| Complex routing & lightweight RPC |
+----------------------+--------------------+--------------------+-----------------------------------+
| Apache Flink         | Stateful Compute   | RocksDB StateStore | Complex streaming event-time joins|
+----------------------+--------------------+--------------------+-----------------------------------+

11. Design Challenge & Exercise

Scenario: Global IoT Log Ingestion Pipeline

You are designing a vehicle telemetry streaming platform handling $2,000,000\text{ messages/sec}$ ($1\text{ GB/sec}$ sustained write ingress). Data must be retained for 3 years for insurance compliance.

Constraints:

• NVMe SSD storage on compute instances costs $0.15/GB/month.
• AWS S3 Standard costs $0.023/GB/month.
• Cold archive S3 Glacier costs $0.0036/GB/month.

Your Task:

• Calculate the 3-year storage bill comparing monolithic Kafka SSD retention versus Pulsar + BookKeeper Tiered Storage.
• Configure the BookKeeper ensemble parameters ($E,W,A$) to withstand the simultaneous loss of 2 bookie storage nodes without data loss.
• Detail the ledger offloading lifecycle rule to transition sealed ledgers to S3 after 4 hours.

12. Assessment

Review the 4 scenario-based quiz questions at the top of this lesson to test your mastery of decoupled messaging architectures, BookKeeper quorum striping, and tiered storage offloading.

13. Summary & Architectural Decision Guide

• Broker / Storage Decoupling: Pulsar stateless brokers scale elastically without rebalancing terabytes of partition disk storage.
• Segmented Ledgers in BookKeeper: Topics are composed of finite, immutable ledgers striped across Bookies, enabling instant partition re-assignment.
• Multi-Tenancy & Tiered Storage: Native tenant/namespace isolation and automatic S3 offloading make Pulsar optimal for enterprise messaging platforms.

14. Authoritative References & Version Context

• Matteo Merli et al. (2018), “Apache Pulsar: Next-Generation Distributed Messaging System”, Apache Software Foundation — pulsar.apache.org/docs/
• Flavio P. Junqueira, Benjamin C. Reed, and Marco Serafini (2013), “Durability with BookKeeper”, ACM Operating Systems Review 47(1) — doi.org/10.1145/2490184.2490190
• Apache BookKeeper Project Documentation (v4.16+) — Ledgers, Quorums, and Journaling — bookkeeper.apache.org/docs/
• Version Context: Architecture reflects Apache Pulsar 3.0+ LTS and Apache BookKeeper 4.16+ enterprise streaming platforms.