KafkaIQ

6.1 The Kafka Controller Internals

Module Overview

Goal: Deep dive into controller behavior — the brain of the Kafka cluster.

What You'll Learn:

  • Controller responsibilities and operations
  • Controller election and failover
  • LeaderAndIsrRequest internals
  • Metadata update propagation
  • Controller state machine
  • Controller bottlenecks and optimizations

What's NOT Covered Here:

  • ZooKeeper/KRaft details (covered in Modules 5.2 and 5.3)
  • Leader election algorithms (covered in Module 4.2)
  • Broker lifecycle (covered in Module 7.1)

Key Insight: The controller is the central coordination point in Kafka. Understanding controller internals explains how Kafka manages partition leadership, handles failures, and maintains cluster state.

Note: This module covers controller behavior in both ZK and KRaft modes, noting differences where applicable.

1. Controller Responsibilities

Core Duties

THE CONTROLLER MANAGES:

1. Partition Leadership:
   ├─ Elect leaders on broker failure
   ├─ Preferred leader election
   └─ Update partition metadata

2. Broker Lifecycle:
   ├─ Detect broker joins/leaves
   ├─ Handle broker registration
   └─ Track broker liveness

3. Topic Operations:
   ├─ Create topics
   ├─ Delete topics
   └─ Modify topic configurations

4. Partition Reassignment:
   ├─ Coordinate replica movement
   ├─ Add/remove replicas
   └─ Update assignments

5. ISR Management:
   ├─ Process ISR change requests from leaders
   ├─ Update cluster metadata
   └─ Notify brokers of changes

6. Metadata Propagation:
   └─ Send UpdateMetadataRequest to all brokers

2. Controller Election

Election Process (ZK Mode)

ZOOKEEPER-BASED ELECTION:

1. All brokers watch /controller znode

2. Controller dies:
   └─ Ephemeral /controller znode deleted

3. All brokers notified (via ZK watch)

4. Brokers race to create /controller:
   ├─ First successful broker becomes controller
   └─ Others continue watching

5. New controller:
   ├─ Increments /controller_epoch
   ├─ Loads cluster state from ZK
   ├─ Sends UpdateMetadataRequest to all brokers
   └─ Starts managing partitions

Election time: ~Seconds (depends on cluster size)

Election Process (KRaft Mode)

KRAFT RAFT-BASED ELECTION:

1. Controller quorum (3-5 controllers)

2. Active controller (Raft leader):
   └─ Handles metadata operations

3. Current leader fails or loses quorum

4. Raft election:
   ├─ Followers detect leader failure
   ├─ Candidate elected via majority votes
   └─ New leader assumes controller role

5. New controller:
   ├─ Continues from metadata log offset
   ├─ No need to reload entire state (already replicated)
   └─ Immediately operational

Election time: ~Milliseconds to seconds (much faster than ZK)

3. Controller State Machine

State Transitions

CONTROLLER STATES:

Idle:
└─ Not the controller (watching for controller election)

ControllerChange:
└─ Just elected, initializing state

NewBrokerChange:
└─ Processing new broker registration

BrokerChange:
└─ Processing broker failure/departure

TopicChange:
└─ Processing topic creation/deletion

PartitionModification:
└─ Processing partition reassignment or ISR change

ControllerShutdown:
└─ Gracefully stepping down as controller

TRANSITIONS:

Idle → ControllerChange:
└─ Elected as controller

ControllerChange → (various states):
└─ Processing events

Any state → Idle:
└─ Lost controller role (failover)

4. Leader Election Coordination

Partition Leader Election

WHEN CONTROLLER ELECTS LEADER:

Trigger:
├─ Broker failure (current leader dead)
├─ Preferred leader election
├─ Partition creation
└─ Controlled shutdown

Process:

1. Controller identifies affected partitions
   └─ All partitions led by failed broker

2. For each partition:
   a. Load current state:
      ├─ Replicas: [1, 2, 3]
      ├─ ISR: [1, 2, 3]
      └─ Leader: 1 (failed)

   b. Select new leader:
      ├─ Remove failed broker from ISR → [2, 3]
      ├─ Choose first alive ISR member → Broker 2
      └─ Increment leader epoch

   c. Update metadata (ZK or metadata log)

3. Send LeaderAndIsrRequest to affected brokers

4. Brokers apply leadership changes

5. Controller sends UpdateMetadataRequest to all brokers

LeaderAndIsrRequest

LEADERANDISR REQUEST STRUCTURE:

LeaderAndIsrRequest:
├─ controller_id: 0
├─ controller_epoch: 5
├─ broker_epoch: 10 (KRaft only)
├─ partition_states:
│   └─ For each affected partition:
│       ├─ topic: "orders"
│       ├─ partition: 0
│       ├─ leader: 2 (new leader)
│       ├─ leader_epoch: 6 (incremented)
│       ├─ isr: [2, 3]
│       ├─ replicas: [1, 2, 3]
│       └─ adding_replicas: [] (if reassignment)
└─ live_leaders: [
      {broker_id: 2, host: "broker2", port: 9092},
      {broker_id: 3, host: "broker3", port: 9092}
    ]

Target brokers: All brokers in replica set

5. ISR Management

Processing ISR Changes

ISR CHANGE FLOW:

1. Partition leader detects follower lag:
   └─ Follower hasn't fetched within replica.lag.time.max.ms

2. Leader updates local ISR:
   └─ ISR: [1, 2, 3] → [1, 2]

3. Leader sends AlterPartitionRequest to controller:
   └─ "Partition orders-0, new ISR: [1, 2], leader epoch: 5"

4. Controller validates request:
   ├─ Check leader epoch matches
   ├─ Check controller epoch matches
   └─ Accept if valid

5. Controller updates metadata:
   ├─ ZK mode: Write to /brokers/topics/{topic}/partitions/{partition}/state
   └─ KRaft mode: Append PartitionChangeRecord to metadata log

6. Controller responds to leader: Success

7. Leader commits ISR change locally

8. Controller sends UpdateMetadataRequest to all brokers
   └─ All brokers learn of new ISR

6. Metadata Propagation

UpdateMetadataRequest

CONTROLLER BROADCASTS METADATA:

When to send:
├─ Leader election
├─ ISR change
├─ Broker join/leave
├─ Topic creation/deletion
└─ Partition reassignment

UpdateMetadataRequest:
├─ controller_id: 0
├─ controller_epoch: 5
├─ live_brokers:
│   └─ [(id=1, host=broker1, port=9092), ...]
├─ partition_states:
│   └─ For each updated partition:
│       ├─ topic: "orders"
│       ├─ partition: 0
│       ├─ leader: 2
│       ├─ isr: [2, 3]
│       └─ replicas: [1, 2, 3]
└─ topic_states: (KRaft only)

Sent to: All brokers in cluster

Brokers apply:
└─ Update local metadata cache
└─ Serve fresh metadata to clients

7. Controller Failover

What Happens When Controller Fails

CONTROLLER FAILOVER:

1. Current controller dies (Broker 0)

2. ZK Mode:
   ├─ Ephemeral /controller znode deleted
   ├─ All brokers notified
   ├─ Brokers race to create /controller
   └─ Winner becomes new controller

   KRaft Mode:
   ├─ Followers detect leader failure (heartbeat timeout)
   ├─ Raft election triggered
   └─ Majority elects new leader (new controller)

3. New controller (Broker 1):
   ├─ Increments controller epoch (5 → 6)
   │
   ├─ ZK Mode:
   │   ├─ Loads all metadata from ZooKeeper
   │   ├─ Scans /brokers/topics/, /brokers/ids/, etc.
   │   └─ Rebuilds in-memory state (slow for large clusters)
   │
   └─ KRaft Mode:
       ├─ Already has metadata log replicated
       ├─ Loads from snapshot + recent log (fast)
       └─ Immediately operational

4. New controller sends UpdateMetadataRequest to all brokers:
   └─ "I'm the new controller (epoch 6)"

5. Brokers accept new controller (higher epoch)

6. Cluster fully operational

Failover time:
├─ ZK mode: Seconds to minutes (large clusters slow)
└─ KRaft mode: Milliseconds to seconds

8. Controller Bottlenecks

Common Performance Issues

CONTROLLER CAN BECOME BOTTLENECK:

1. Large Partition Count:
   ├─ Millions of partitions
   ├─ ZK mode: Must scan all metadata (slow failover)
   └─ KRaft mode: Better, but still high memory usage

2. Frequent ISR Changes:
   ├─ High rate of AlterPartitionRequests
   ├─ Controller processes sequentially
   └─ Can saturate controller CPU

3. Broker Failures:
   ├─ Controller must elect leaders for all affected partitions
   ├─ Large clusters: Thousands of partitions per broker
   └─ Can take minutes to complete

4. ZooKeeper Latency:
   ├─ Every metadata change writes to ZK
   ├─ ZK write latency compounds
   └─ Slows down controller operations

5. Single-threaded Processing (ZK mode):
   └─ Controller event queue processes one event at a time
   └─ Can fall behind under high load

Optimizations (KRaft)

KRAFT IMPROVEMENTS:

1. Batched Metadata Writes:
   └─ Multiple changes in single log append
   └─ Reduces round-trips

2. Snapshots:
   └─ Fast failover (snapshot + recent log)
   └─ No need to scan all metadata

3. Parallel Processing:
   └─ Better concurrency for metadata operations

4. Lower Latency:
   └─ No external ZK calls (in-memory + local log)

5. Higher Partition Scale:
   └─ Tested with millions of partitions

9. Controller Operations

Topic Creation

CREATE TOPIC FLOW:

1. Client sends CreateTopicsRequest to any broker

2. Broker forwards to controller

3. Controller:
   a. Validates request (name, partition count, RF)
   b. Selects replica assignments:
      └─ Distribute partitions across brokers
   c. Writes metadata:
      └─ ZK: Create znodes
      └─ KRaft: Append TopicRecord, PartitionRecords
   d. Sends LeaderAndIsrRequest to brokers hosting replicas

4. Brokers:
   ├─ Create partition directories
   ├─ Initialize logs
   └─ Start replication (if follower)

5. Controller sends UpdateMetadataRequest to all brokers

6. Topic available for produce/consume

Partition Reassignment

PARTITION REASSIGNMENT FLOW:

Goal: Move partition from [1, 2, 3] to [2, 3, 4]

1. Client submits reassignment request to controller

2. Controller:
   a. Add new replica (Broker 4):
      └─ Replicas: [1, 2, 3, 4]
      └─ Send LeaderAndIsrRequest

   b. Broker 4 starts replicating (full sync)

   c. Wait for Broker 4 to catch up (join ISR)

   d. If leader is being removed (Broker 1):
      └─ Elect new leader from [2, 3, 4]

   e. Remove old replica (Broker 1):
      └─ Replicas: [2, 3, 4]
      └─ Send StopReplicaRequest to Broker 1

   f. Update metadata

3. Reassignment complete
   └─ Partition now on [2, 3, 4]

10. Controller Observability: Log Messages

Understanding Controller Logs

CONTROLLER LOGS:

Purpose:
└─ Track controller state changes and operations
└─ Essential for debugging and monitoring
└─ Indicates cluster health

Location:
└─ Kafka log files (server.log or controller.log)
└─ Typically: /var/log/kafka/server.log

Key Log Messages to Monitor

CONTROLLER ELECTION:

Successful election:
INFO [Controller] Successfully elected as the controller (kafka.controller.KafkaController)
INFO [ControllerEventThread] Starting controller event processor thread (kafka.controller.ControllerEventManager)
INFO [Controller] Elected as the controller with epoch 5

What it means:
├─ This broker became the active controller
├─ Controller epoch incremented
└─ Ready to process events

When you see it:
├─ Broker startup (if elected)
├─ Previous controller failed
└─ Manual controller failover

CONTROLLER STEPPING DOWN:

Relinquishing leadership:
INFO [Controller] Relinquishing controller leadership (kafka.controller.KafkaController)
INFO [ControllerEventThread] Shutting down controller event processor thread (kafka.controller.ControllerEventManager)
INFO [Controller] Resigned as the controller at epoch 5

What it means:
├─ This broker no longer controller
├─ Graceful shutdown or failover
└─ New election will occur

When you see it:
├─ Broker shutdown (controlled)
├─ Controller explicitly stepped down
└─ ZK session expired / KRaft leadership lost

CONTROLLER PROCESSING EVENTS:

Event processing:
INFO [Controller] Processing BrokerChange event (kafka.controller.ControllerEventManager)
INFO [Controller] New broker 3 registered (kafka.controller.KafkaController)
INFO [Controller] Broker 3 started (kafka.controller.KafkaController)
INFO [Controller] Removed broker 2 from metadata (kafka.controller.KafkaController)

What it means:
├─ Controller handling cluster changes
├─ Broker registration/deregistration
└─ Normal cluster operations

Event types:
├─ BrokerChange: Broker joined/left
├─ TopicChange: Topic created/deleted
├─ PartitionModification: Partition reassignment
├─ IsrChange: ISR membership changed
└─ ControllerChange: New controller elected

CONTROLLER EPOCH CHANGE:

Epoch increment:
INFO [Controller] Updated controller epoch to 5 (kafka.controller.KafkaController)
INFO [Controller] Incremented controller epoch to 6 in ZooKeeper (kafka.controller.KafkaController)

What it means:
├─ New controller elected
├─ Epoch prevents stale controller actions
└─ Monotonically increasing

Significance:
└─ Each controller generation has unique epoch
└─ Brokers reject requests from old epochs

PARTITION LEADERSHIP CHANGES:

Leader election:
INFO [Controller] Starting preferred leader election for partitions (kafka.controller.KafkaController)
INFO [Controller] Elected leader 2 for partition orders-0 at offset 1234 (kafka.controller.KafkaController)
INFO [Controller] Successfully elected leader for 10 partitions (kafka.controller.KafkaController)

What it means:
├─ Controller elected new partition leaders
├─ After broker failure or rebalance
└─ Partitions now available

When you see it:
├─ Broker failure
├─ Preferred leader election
└─ Partition reassignment

ISR UPDATES:

ISR changes:
INFO [Controller] Partition [orders,0] ISR updated to [1,2] (kafka.controller.KafkaController)
WARN [Controller] Partition [orders,0] ISR shrunk to [1] (kafka.controller.KafkaController)
INFO [Controller] Partition [orders,0] ISR expanded to [1,2,3] (kafka.controller.KafkaController)

What it means:
├─ ISR membership changed
├─ Follower fell behind or caught up
└─ Impacts durability guarantees

Watch for:
├─ Frequent ISR shrinks: Performance issues
├─ ISR size = 1: Reduced fault tolerance
└─ ISR never expanding: Replication problems

Error and Warning Messages

CONTROLLER ERRORS:

Connection issues:
ERROR [Controller] Error processing event BrokerChange (kafka.controller.ControllerEventManager)
ERROR [Controller] Failed to send LeaderAndIsrRequest to broker 2 (kafka.controller.KafkaController)
WARN [Controller] Broker 2 is not available, skipping LeaderAndIsrRequest (kafka.controller.KafkaController)

What it means:
├─ Controller communication failure
├─ Broker unreachable
└─ May impact metadata propagation

Action required:
└─ Investigate broker health
└─ Check network connectivity

ELECTION FAILURES:

Election problems:
ERROR [Controller] Controller election failed (kafka.controller.KafkaController)
WARN [Controller] Multiple brokers attempting to become controller (kafka.controller.KafkaController)
ERROR [Controller] ZooKeeper session expired during controller election (kafka.controller.KafkaController)

What it means:
├─ Controller election unsuccessful
├─ ZK/KRaft issues
└─ Cluster may be in degraded state

Action required:
└─ Check ZooKeeper/KRaft health
└─ Verify network connectivity
└─ Review controller logs on all brokers

Monitoring Tips

HEALTHY CONTROLLER LOGS:

Normal operation:
├─ Controller elected once (startup)
├─ Event processing completes quickly (<100ms)
├─ Few ISR changes (stable replication)
└─ No repeated errors

Patterns to watch:

✅ Good:
INFO [Controller] Successfully elected as the controller
INFO [Controller] Processing BrokerChange event
INFO [Controller] Elected leader for partition orders-0

❌ Bad (indicates issues):
ERROR [Controller] Error processing event
WARN [Controller] Controller flapping between brokers
ERROR [Controller] Failed to send LeaderAndIsrRequest repeatedly
WARN [Controller] ISR shrunk to 1 for multiple partitions

Log Analysis Commands

# Find controller elections
grep "Successfully elected as the controller" /var/log/kafka/server.log

# Count elections (should be low)
grep -c "Successfully elected as the controller" /var/log/kafka/server.log

# Find controller errors
grep ERROR /var/log/kafka/server.log | grep Controller

# Monitor ISR changes
grep "ISR updated" /var/log/kafka/server.log | tail -20

# Check controller epoch
grep "controller epoch" /var/log/kafka/server.log | tail -5

# Find broker failures (controller perspective)
grep "Removed broker" /var/log/kafka/server.log

# Monitor event processing time
grep "Processing.*event" /var/log/kafka/server.log | \
  awk '{print $NF}' | sort -n

Alerting on Controller Issues

RECOMMENDED ALERTS:

1. Controller flapping:
   └─ More than 5 elections in 10 minutes
   └─ Indicates: Network issues, ZK problems, overload

2. Event processing slow:
   └─ Events taking >1 second to process
   └─ Indicates: Controller overwhelmed, hardware issue

3. Failed metadata propagation:
   └─ LeaderAndIsrRequest failures
   └─ Indicates: Broker unreachable, network issues

4. ISR shrink storm:
   └─ Multiple partitions losing ISR members
   └─ Indicates: Broker performance issues, network problems

Example alert script:
```bash
#!/bin/bash
# Alert on controller flapping

ELECTIONS=$(grep -c "Successfully elected as the controller" \
  /var/log/kafka/server.log | \
  tail -100)

if [ "$ELECTIONS" -gt 5 ]; then
  echo "ALERT: Controller flapping detected ($ELECTIONS elections)"
  # Send to monitoring system
  exit 1
fi


---

### Log Levels for Controller

CONFIGURING LOG LEVELS:

In log4j.properties:

Default (INFO level)

log4j.logger.kafka.controller=INFO

Debug controller operations

log4j.logger.kafka.controller=DEBUG

Trace all controller events (very verbose)

log4j.logger.kafka.controller=TRACE

Only errors and warnings

log4j.logger.kafka.controller=WARN

Recommendation: ├─ Production: INFO (default) ├─ Troubleshooting: DEBUG (temporarily) └─ Never use TRACE in production (too verbose)

Dynamic log level change (no restart):

kafka-log-dirs.sh --describe \
  --log-level kafka.controller=DEBUG


---

### Summary: Log Message Patterns

CONTROLLER LIFECYCLE LOGS:

Startup → Election → Processing Events → Shutdown

  1. Broker starts: "Starting Kafka server"

  2. Controller election: "Successfully elected as the controller"

  3. Normal operation: "Processing BrokerChange event" "Elected leader for partition X" "ISR updated for partition Y"

  4. Graceful shutdown: "Relinquishing controller leadership" "Shutting down controller event processor"

ERROR PATTERNS TO INVESTIGATE:

  • Repeated election messages (flapping)
  • LeaderAndIsrRequest failures
  • Event processing errors
  • ZooKeeper/KRaft connection issues
  • Timeout errors

HEALTHY CLUSTER: └─ Controller elected once per broker lifetime └─ Events process quickly └─ Few ISR changes └─ No repeated errors


---

## 11. Key Takeaways

**You've learned:**

✅ **Controller is the cluster brain** — manages leadership, failures, metadata  
✅ **Election:** ZK (race to create znode) vs KRaft (Raft leader)  
✅ **LeaderAndIsrRequest** sent to brokers on leadership changes  
✅ **AlterPartitionRequest** from leaders to controller for ISR changes  
✅ **UpdateMetadataRequest** broadcasts metadata to all brokers  
✅ **Controller failover:** ZK (slow, load from ZK) vs KRaft (fast, snapshot + log)  
✅ **Controller bottlenecks:** Large clusters, ISR churn, ZK latency  
✅ **KRaft optimizations:** Batching, snapshots, parallelism  
✅ **Topic creation:** Controller assigns replicas, brokers initialize partitions  
✅ **Partition reassignment:** Add new replica, sync, remove old replica

---

## Summary

**The controller is Kafka's central coordinator:**

Responsibilities = ├─ Partition leadership (elect leaders, track ISR) ├─ Broker lifecycle (detect joins/failures) ├─ Topic operations (create, delete, configure) ├─ Partition reassignment (move replicas) └─ Metadata propagation (UpdateMetadataRequest)

Key Requests = ├─ LeaderAndIsrRequest → Brokers (leadership changes) ├─ AlterPartitionRequest → Controller (ISR updates) ├─ UpdateMetadataRequest → All brokers (metadata sync) └─ StopReplicaRequest → Brokers (remove replicas)

Election = ├─ ZK: Race to create /controller znode └─ KRaft: Raft leader election

Failover = ├─ ZK: Slow (load all metadata from ZK) └─ KRaft: Fast (snapshot + recent log)

Bottlenecks = ├─ Large partition count ├─ Frequent ISR changes ├─ ZK latency (ZK mode) └─ Mitigated by KRaft


**Remember:** The controller is critical for cluster health. Understanding controller internals explains Kafka's coordination behavior and guides capacity planning (especially in large clusters).

**Next up:** Module 7.1 — Broker Lifecycle & State Transitions — Internal broker state machine.