KafkaIQ

7.1 Broker Lifecycle & State Transitions

Module Overview

Goal: Understand the internal broker state machine and lifecycle phases.

What You'll Learn:

  • Broker startup phases and initialization
  • Registration with controller
  • Partition loading and recovery
  • Controlled shutdown internals
  • Replica state transitions
  • Broker state machine

What's NOT Covered Here:

  • Controller operations (covered in Module 6.1)
  • Partition internals (covered in Module 2.x)
  • Replication protocol (covered in Module 3.x)

Key Insight: Brokers go through well-defined phases during startup, operation, and shutdown. Understanding broker lifecycle explains startup behavior, recovery scenarios, and graceful shutdown mechanics.

1. Broker Startup Sequence

High-Level Phases

BROKER STARTUP:

1. Configuration Loading
   └─ Parse server.properties

2. Log Manager Initialization
   └─ Load partitions from disk

3. Coordination Registration
   └─ Register with ZooKeeper or KRaft

4. Network Layer Startup
   └─ Start accepting client connections

5. Controller Notification
   └─ Controller assigns leadership

6. Fully Operational
   └─ Ready to serve traffic

Typical startup time:
├─ Small cluster: Seconds
├─ Large cluster (many partitions): Minutes
└─ After crash: Slower (log recovery)

Phase 1: Configuration Loading

LOAD CONFIGURATION:

1. Read server.properties:
   ├─ broker.id or node.id
   ├─ listeners (PLAINTEXT://0.0.0.0:9092)
   ├─ log.dirs (/var/kafka/data)
   ├─ zookeeper.connect or controller.quorum.voters
   └─ All other configs

2. Validate configuration:
   ├─ Required configs present
   ├─ No conflicting settings
   └─ Directories writable

3. Initialize logging framework (Log4j)

4. Display startup banner (version info)

Phase 2: Log Manager Initialization

LOAD PARTITIONS FROM DISK:

1. Scan log.dirs:
   └─ /var/kafka/data/, /var/kafka/data2/, ...

2. For each partition directory (e.g., orders-0/):
   a. Load partition metadata
   b. Scan log segments:
      ├─ 00000000000000000000.log
      ├─ 00000000000001000000.log
      └─ ... (all segments)
   c. Validate integrity:
      ├─ Check CRCs
      ├─ Verify offset sequence
      └─ Detect corruption
   d. Rebuild indexes (if missing or incomplete):
      ├─ .index files
      └─ .timeindex files
   e. Recover active segment (if unclean shutdown):
      └─ Truncate incomplete records
   f. Load leader epoch checkpoint
   g. Create in-memory Log object

3. Initialize log cleaner threads (for compaction)

4. Mark log manager as started

Time: Depends on partition count and log size
└─ Large partitions: Minutes to recover

Phase 3: Registration with Controller

ZK MODE REGISTRATION:

1. Connect to ZooKeeper ensemble

2. Create ephemeral znode:
   └─ /brokers/ids/{broker_id}
   └─ Data: {"version": 4, "host": "broker1", "port": 9092, ...}

3. Controller detects new broker (via watch)

4. Controller sends UpdateMetadataRequest
   └─ Broker receives cluster metadata

5. Controller sends LeaderAndIsrRequest
   └─ Broker learns partition assignments

KRAFT MODE REGISTRATION:

1. Generate or load broker UUID

2. Send BrokerRegistrationRequest to controller:
   └─ broker_id, UUID, host, port, listeners, rack

3. Controller appends RegisterBrokerRecord to metadata log

4. Controller responds: Registration successful

5. Broker starts heartbeat loop:
   └─ Periodic BrokerHeartbeatRequest to controller

6. Broker replicates metadata log (as observer)

7. Broker learns partition assignments from metadata log

Phase 4: Network Layer Startup

START ACCEPTING CONNECTIONS:

1. Create Acceptor thread:
   └─ Bind to configured listeners (e.g., 0.0.0.0:9092)

2. Create Processor threads (num.network.threads):
   └─ Read/write from client connections

3. Create Request Handler threads (num.io.threads):
   └─ Process client requests

4. Acceptor starts accepting connections

5. Broker now reachable by clients

Note: Broker may not be fully ready:
└─ Partitions still loading or recovering
└─ Leadership not yet assigned

Phase 5: Controller Notification & Leadership Assignment

CONTROLLER ASSIGNS PARTITIONS:

1. Controller detects new broker

2. Controller checks partition assignments:
   └─ Which partitions have replicas on this broker?

3. For each partition:
   a. If broker should be follower:
      └─ Send LeaderAndIsrRequest with follower role

   b. If broker should be leader (preferred leader):
      └─ Send LeaderAndIsrRequest with leader role
      └─ (Or wait for preferred leader election)

4. Broker processes LeaderAndIsrRequest:
   ├─ For leader partitions: Start accepting writes
   ├─ For follower partitions: Start replicating
   └─ Update local metadata

5. Broker fully operational

2. Partition Loading

Loading Replicas

FOR EACH PARTITION ON BROKER:

1. Create Replica object:
   ├─ Topic: orders
   ├─ Partition: 0
   ├─ Log: Reference to Log object (loaded earlier)
   └─ State: NewReplica

2. Wait for controller assignment:
   └─ LeaderAndIsrRequest determines role

3. Transition to operational state:
   ├─ OnlineReplica (follower)
   └─ Or leader (if assigned)

4. Start serving:
   ├─ Leader: Accept produce/fetch from clients
   └─ Follower: Fetch from leader

3. Controlled Shutdown

Graceful Shutdown Process

CONTROLLED SHUTDOWN:

1. Broker receives shutdown signal (SIGTERM)

2. Broker sends ControlledShutdownRequest to controller:
   └─ "I'm shutting down, move my partitions"

3. Controller processes:
   a. Identify partitions led by shutting-down broker
   b. For each partition:
      ├─ Elect new leader from ISR (excluding this broker)
      ├─ Send LeaderAndIsrRequest to new leader
      └─ New leader starts accepting writes
   c. Respond to broker: "Leadership transferred"

4. Broker completes shutdown:
   a. Stop accepting new client connections
   b. Wait for in-flight requests to complete
   c. Flush logs to disk:
      └─ Ensure no data loss
   d. Close network layer
   e. Shutdown log manager
   f. Disconnect from ZK/KRaft
   g. Exit process

Result:
└─ Zero downtime (leadership transferred before shutdown)
└─ Clean state (logs flushed, no recovery needed on restart)

Controlled shutdown timeout:
└─ controlled.shutdown.max.retries (default: 3)
└─ controlled.shutdown.retry.backoff.ms (default: 5000ms)

vs Uncontrolled Shutdown (Crash)

CRASH SCENARIO:

1. Broker dies suddenly (kill -9, power loss, etc.)

2. No ControlledShutdownRequest sent

3. ZK Mode:
   ├─ ZK session expires (after zookeeper.session.timeout.ms)
   ├─ Ephemeral znode deleted
   └─ Controller detects failure

   KRaft Mode:
   ├─ Heartbeat timeout
   └─ Controller marks broker offline

4. Controller reacts:
   ├─ Elect new leaders for affected partitions
   └─ Send LeaderAndIsrRequest to new leaders

5. Brief client unavailability:
   └─ Until new leaders elected and metadata propagated

6. Broker restart:
   ├─ Longer recovery (validate logs, rebuild indexes)
   ├─ Active segment may have incomplete records (truncated)
   └─ Divergent data truncated (via leader epochs)

Impact:
└─ Brief downtime (seconds)
└─ Longer restart time

4. Replica State Machine

Replica States

REPLICA STATE TRANSITIONS:

NewReplica:
└─ Replica created, not yet started

OnlineFetchingReplica (Follower):
├─ Actively replicating from leader
└─ Normal operational state

OnlineReplica (Leader):
├─ Accepting client writes
└─ Serving client reads

OfflineReplica:
└─ Broker hosting replica is down

TruncatingLog:
└─ Truncating divergent log after leader change

DeletingReplica:
└─ Replica being deleted (partition reassignment or topic deletion)

NonExistentReplica:
└─ Replica fully deleted

State Transition Diagram

STATE TRANSITIONS:

NonExistentReplica → NewReplica:
└─ Partition created or reassignment adds replica

NewReplica → OnlineFetchingReplica:
└─ Broker receives LeaderAndIsrRequest (follower role)

NewReplica → OnlineReplica (Leader):
└─ Broker receives LeaderAndIsrRequest (leader role)

OnlineFetchingReplica → OnlineReplica (Leader):
└─ Elected as leader (failover)

OnlineReplica (Leader) → OnlineFetchingReplica:
└─ Demoted to follower (leader election elsewhere)

OnlineReplica/OnlineFetchingReplica → OfflineReplica:
└─ Broker failure

OfflineReplica → TruncatingLog:
└─ Broker restarts, detects divergence

TruncatingLog → OnlineFetchingReplica:
└─ Truncation complete, resume replication

OnlineReplica/OnlineFetchingReplica → DeletingReplica:
└─ Partition deletion or reassignment removal

DeletingReplica → NonExistentReplica:
└─ Replica fully deleted

5. Log Recovery

Recovery After Unclean Shutdown

ACTIVE SEGMENT RECOVERY:

1. Broker starts, detects unclean shutdown (flag file)

2. For each partition's active segment:
   a. Read records sequentially
   b. Validate each record batch:
      ├─ Check CRC
      ├─ Verify offset sequence
      └─ Ensure completeness
   c. If incomplete or corrupted record found:
      ├─ Truncate at last valid offset
      └─ Discard partial write
   d. Update LEO to last valid offset

3. Rebuild indexes:
   ├─ Scan valid records
   ├─ Regenerate .index and .timeindex
   └─ Write index files to disk

4. Mark partition as recovered

5. Continue startup

Recovery time:
└─ Depends on active segment size
└─ Large segments: Minutes

6. Broker Heartbeats (KRaft Mode)

Heartbeat Protocol

KRAFT BROKER HEARTBEAT:

1. Broker sends BrokerHeartbeatRequest to controller:
   └─ Every broker.heartbeat.interval.ms (default: 2 seconds)

2. Request includes:
   ├─ broker_id
   ├─ broker_epoch
   ├─ current_metadata_offset (log offset)
   └─ want_fence (false, unless fencing self)

3. Controller responds:
   ├─ is_caught_up: true/false (metadata log replication)
   ├─ is_fenced: false (broker allowed to operate)
   └─ should_shutdown: false

4. If heartbeat missed (timeout > session.timeout.ms):
   └─ Controller marks broker offline
   └─ Triggers partition leader elections

5. Broker continues heartbeat loop until shutdown

Purpose:
└─ Prove liveness to controller
└─ Sync metadata log replication status

7. Broker Startup Observability

Monitoring Broker Startup

WHY MONITOR STARTUP:

Diagnose:
├─ Startup failures
├─ Slow recovery
├─ Configuration issues
└─ Partition loading problems

Track:
├─ Startup duration
├─ Partition loading progress
├─ Controller election
└─ Readiness status

Key Log Messages During Startup

PHASE 1: CONFIGURATION LOADING

Log messages:
INFO [KafkaServer] Starting Kafka version 3.x.x
INFO [KafkaServer] Kafka server started with broker ID 1
INFO [KafkaServer] Registered broker 1 in cluster with ZK/KRaft

What it indicates:
├─ Broker process started successfully
├─ Configuration loaded and validated
└─ Broker ID assigned/confirmed

Look for errors:
ERROR [KafkaServer] Fatal error during broker startup
└─ Configuration invalid, exiting

PHASE 2: LOG MANAGER INITIALIZATION

Log messages:
INFO [LogManager] Loading logs from /var/lib/kafka/data
INFO [Log partition=orders-0, dir=/var/lib/kafka/data] Completed load of log with 3 segments, size 1024 MB
INFO [Log partition=users-1, dir=/var/lib/kafka/data] Recovering segment offsets 0-1000000
INFO [LogManager] Loaded 1234 logs in 3456 ms
INFO [LogManager] Logs loaded successfully

What it indicates:
├─ Partition logs discovered
├─ Segments validated and loaded
├─ Recovery completed (if needed)
└─ Time taken for log loading

Performance indicator:
└─ Large partition count or corrupted logs → slow startup

PHASE 3: CLUSTER REGISTRATION

ZK Mode:
INFO [KafkaServer] Connecting to ZooKeeper at zk1:2181,zk2:2181,zk3:2181
INFO [KafkaServer] Successfully connected to ZooKeeper
INFO [KafkaServer] Registered broker 1 in ZooKeeper

KRaft Mode:
INFO [KafkaServer] Registered broker 1 with controller quorum
INFO [BrokerServer] Completed registration with controller
INFO [MetadataLoader] Loaded metadata log up to offset 5678

What it indicates:
├─ Broker joined cluster successfully
├─ Metadata synchronization started
└─ Ready to receive partition assignments

PHASE 4: CONTROLLER STATUS

Controller election (if applicable):
INFO [Controller] Controller moved to broker 1
INFO [Controller] Elected as the controller with epoch 5
INFO [ControllerEventThread] Starting controller event processor thread

Non-controller broker:
INFO [MetadataCache] Updated metadata for cluster
INFO [Replica Manager] Started replication for partitions [orders-0, users-1, ...]

What it indicates:
├─ Broker knows its role (controller or data broker)
├─ Partition assignments received
└─ Ready for traffic

PHASE 5: READY FOR TRAFFIC

Final startup messages:
INFO [KafkaServer] Kafka broker 1 is ready to serve requests
INFO [SocketServer] Accepting connections on listener PLAINTEXT://0.0.0.0:9092
INFO [Kafka Server] started (kafka.server.KafkaServer)

What it indicates:
├─ All phases complete
├─ Network layer accepting connections
└─ Broker fully operational

Internal Topics to Monitor

KAFKA INTERNAL TOPICS:

1. __cluster_metadata (KRaft mode):
```bash
kafka-topics.sh --bootstrap-server localhost:9092 --list | grep metadata
# Output: __cluster_metadata

kafka-topics.sh --bootstrap-server localhost:9092 \
  --describe --topic __cluster_metadata
# Should show: 1 partition, RF=3 (or quorum size)

Purpose: └─ Stores cluster metadata in KRaft mode └─ Presence indicates KRaft operational

  1. __consumer_offsets:
kafka-topics.sh --bootstrap-server localhost:9092 --list | grep offsets
# Output: __consumer_offsets

kafka-topics.sh --bootstrap-server localhost:9092 \
  --describe --topic __consumer_offsets
# Should show: 50 partitions (default), RF=3

Purpose: └─ Stores consumer group offsets └─ Created automatically on first consumer group

  1. __transaction_state:
kafka-topics.sh --bootstrap-server localhost:9092 --list | grep transaction
# Output: __transaction_state

# Should show: 50 partitions (default), RF=3

Purpose: └─ Stores transactional producer state └─ Created automatically on first transaction


---

### JMX Metrics for Startup

BROKER STATE METRIC:

kafka.server:type=KafkaServer,name=BrokerState

Values: ├─ 0: Not Running ├─ 1: Starting ├─ 2: RecoveringFromUncleanShutdown ├─ 3: RunningAsBroker ├─ 4: RunningAsController ├─ 5: PendingControlledShutdown └─ 6: BrokerShuttingDown

Monitor during startup:

# Using jmxterm
echo "get -b kafka.server:type=KafkaServer,name=BrokerState Value" | \
  java -jar jmxterm.jar -l localhost:9999 -n

# Should transition: 0 → 1 → 3 (or 4 if controller)

Expected transition time: ├─ Small cluster: 0→3 in 5-10 seconds ├─ Large cluster: 0→3 in 30-60 seconds └─ Recovery: 2→3 may take minutes (validate logs)


---

LOG LOADING METRICS:

kafka.log:type=LogManager,name=LogLoadTime └─ Time taken to load all partition logs

kafka.log:type=LogManager,name=NumPartitions └─ Total partitions loaded on this broker

Monitor:

echo "get -b kafka.log:type=LogManager,name=LogLoadTime Value" | \
  java -jar jmxterm.jar -l localhost:9999 -n

Alert if: └─ LogLoadTime > 5 minutes (investigate slow recovery)


---

REPLICATION METRICS:

kafka.server:type=ReplicaManager,name=LeaderCount └─ Number of partitions this broker leads

kafka.server:type=ReplicaManager,name=PartitionCount └─ Total partitions (leader + follower) on this broker

Monitor after startup:

# Should match expected partition assignments
echo "get -b kafka.server:type=ReplicaManager,name=PartitionCount Value" | \
  java -jar jmxterm.jar -l localhost:9999 -n


---

## 8. Common Broker Startup Pitfalls

### Troubleshooting Table

COMMON ISSUES AND SOLUTIONS:


| Issue | Cause | Solution |
|-------|-------|----------|
| **Stuck on log recovery** | Corrupted segment or disk issue | Check disk health with `smartctl`, delete corrupt segment, restore from replica |
| **Broker not joining cluster** | Misconfigured `broker.id` or `node.id`, network issues | Verify ID uniqueness: `grep broker.id config/*.properties`, check network: `telnet broker1 9092` |
| **Controller flapping** | ZooKeeper instability or high load | Increase ZK session timeout: `zookeeper.session.timeout.ms=18000`, check ZK health |
| **Long startup time** | Too many partitions on broker | Reduce partitions per broker, increase `num.recovery.threads.per.data.dir=4` |
| **Out of memory** | Large partition count, insufficient heap | Increase JVM heap: `KAFKA_HEAP_OPTS="-Xmx4g -Xms4g"`, reduce partition count |
| **Port already in use** | Previous broker process still running | Kill old process: `kill -9 $(lsof -t -i:9092)` or `pkill -9 kafka` |
| **Cannot connect to ZK/KRaft** | Network issue, ZK/controller down | Verify ZK/controller health, check firewall rules, test connectivity |
| **Metadata not syncing** | Controller not reachable | Check controller logs, verify controller election, test network to controller |

---

### Detailed Troubleshooting Steps

ISSUE: Stuck on Log Recovery

Symptoms: ├─ Broker startup hangs ├─ Log shows: "Recovering segment offsets..." └─ No progress for >5 minutes

Diagnosis:

# Check disk health
sudo smartctl -a /dev/sda

# Check for corrupted segments
grep ERROR /var/log/kafka/server.log | grep -i corrupt

# Check disk I/O
iostat -x 1 10

Solution:

# If specific segment corrupted, delete it
rm /var/lib/kafka/data/orders-0/00000000000012345678.log

# Broker will recover from replica

# If disk failing, replace disk and restore from replica


---

ISSUE: Broker Not Joining Cluster

Symptoms: ├─ Broker starts but not listed in metadata ├─ Producers/consumers cannot connect └─ kafka-broker-api-versions.sh fails

Diagnosis:

# Verify broker.id uniqueness
grep broker.id /etc/kafka/server.properties
kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic __cluster_metadata

# Check network connectivity
telnet broker1.example.com 9092

# Check firewall
sudo iptables -L -n | grep 9092

Solution:

# Fix broker.id if duplicate
broker.id=4  # Ensure unique

# Fix network/firewall
sudo iptables -A INPUT -p tcp --dport 9092 -j ACCEPT

# Restart broker
kafka-server-stop.sh && kafka-server-start.sh config/server.properties


---

ISSUE: Controller Flapping

Symptoms: ├─ Controller election every few seconds ├─ Logs show: "Elected as controller" repeatedly └─ Cluster unstable

Diagnosis:

# Count controller elections
grep -c "Successfully elected as the controller" /var/log/kafka/server.log
# If > 10 in last hour, flapping

# Check ZK health (if ZK mode)
echo stat | nc zookeeper1 2181

# Check network latency
ping -c 10 controller1.example.com

Solution:

# Increase ZK session timeout
zookeeper.session.timeout.ms=18000  # 18 seconds (from default 6s)

# Check ZK capacity
# Increase ZK heap if needed: -Xmx2g

# Reduce broker load
# Scale cluster if overwhelmed


---

ISSUE: Long Startup Time

Symptoms: ├─ Broker takes >10 minutes to start ├─ Log shows slow partition loading └─ "Loaded 5000 logs in 600000 ms"

Diagnosis:

# Check partition count
kafka-topics.sh --bootstrap-server localhost:9092 --describe | wc -l

# Check disk I/O
iostat -x 1 10
# Look for high %util

# Check recovery threads
grep num.recovery.threads.per.data.dir config/server.properties

Solution:

# Increase recovery threads
num.recovery.threads.per.data.dir=4  # Default: 1

# Use faster disks (SSDs)
# Or spread partitions across more brokers

# Reduce partition count per broker
# via partition reassignment


---

### Monitoring Startup Progress

```bash
# Watch broker state in real-time
watch -n 1 'echo "get -b kafka.server:type=KafkaServer,name=BrokerState Value" | java -jar jmxterm.jar -l localhost:9999 -n'

# Monitor log loading
tail -f /var/log/kafka/server.log | grep "Loading logs"

# Check partition count
watch -n 1 'kafka-topics.sh --bootstrap-server localhost:9092 --describe | grep "Leader: 1" | wc -l'

# Verify broker registered
kafka-broker-api-versions.sh --bootstrap-server localhost:9092 | grep "broker 1"

9. Key Takeaways

You've learned:

✅ Broker startup — config load, log recovery, registration, network start
✅ Log manager initialization — scan partitions, validate, rebuild indexes
✅ Registration: ZK (ephemeral znode) vs KRaft (BrokerRegistrationRequest)
✅ Controlled shutdown — proactive leadership transfer, zero downtime
✅ Uncontrolled shutdown (crash) — controller detects timeout, elects new leaders
✅ Replica states — New → Online → Offline → Truncating → Deleting
✅ Log recovery — truncate incomplete records, rebuild indexes
✅ KRaft heartbeats — prove liveness, sync metadata log
✅ Startup time — depends on partition count and log size
✅ Graceful shutdown — flush logs, transfer leadership, clean exit

Summary

Broker lifecycle is well-structured and fault-tolerant:

Startup Phases =
├─ 1. Load configuration
├─ 2. Initialize log manager (load partitions)
├─ 3. Register with controller (ZK or KRaft)
├─ 4. Start network layer
├─ 5. Receive partition assignments
└─ 6. Fully operational

Shutdown Types =
├─ Controlled: ControlledShutdownRequest, leadership transfer, clean exit
└─ Uncontrolled (crash): Timeout detection, controller reacts

Replica States =
├─ New → Online (Leader/Follower)
├─ Online → Offline (failure)
├─ Offline → Truncating → Online (recovery)
└─ Online → Deleting → NonExistent (deletion)

Recovery =
├─ Validate active segment
├─ Truncate incomplete records
├─ Rebuild indexes
└─ Resume operation

Liveness =
├─ ZK: Ephemeral znode (session timeout)
└─ KRaft: Heartbeat loop (BrokerHeartbeatRequest)

Remember: Understanding broker lifecycle explains startup times, recovery behavior, and shutdown mechanics. Controlled shutdown is always preferable to crashes (faster recovery, no data loss).

Next up: Module 8.1 — Transaction & Exactly-Once Internals (Broker Side) — The final module!