Managing Cluster Operations

1. Adding Brokers to Cluster

Property	Description	Default Value	Notes
broker.id	Unique identifier for the broker within the cluster	-1	Must be unique across all brokers in the cluster
listeners	Network listeners for the broker to accept connections	PLAINTEXT://localhost:9092	Can include multiple listeners (e.g., PLAINTEXT, SSL)
advertised.listeners	Address advertised to clients for connecting to the broker	Same as listeners	Must be accessible from client machines
log.dirs	Directories where log segments are stored	/tmp/kafka-logs	Can be a comma-separated list of paths
num.partitions	Default number of partitions for new topics	1	Can be overridden per topic

Example: Adding a New Broker Configuration

broker.id=2
listeners=PLAINTEXT://kafka-broker-2:9092
advertised.listeners=PLAINTEXT://kafka-broker-2:9092
log.dirs=/var/lib/kafka/data
num.partitions=3

Note: When adding a new broker, ensure that the broker.id is unique and that all brokers in the cluster can communicate with each other.

Workflow: Adding a Broker to an Existing Cluster

Stop the broker process if it's already running
Update the configuration file with a unique broker.id
Ensure all brokers can reach the new broker via listeners
Start the new broker with updated configuration
Verify that the new broker joins the cluster using kafka-topics.sh

2. Removing Brokers from Cluster

Removing brokers from a Kafka cluster is a critical operation that requires careful planning to maintain data consistency and cluster stability. The process involves decommissioning brokers gracefully to ensure that all partitions are redistributed across remaining brokers.

Step	Description	Action
1	Identify brokers to remove	Use `kafka-broker-list.sh` or check cluster metadata to identify the broker IDs to decommission.
2	Update broker configuration	Add `broker.id` to the configuration file and set `unclean.leader.election.enable` to `false` for safety.
3	Start decommissioning	Run `kafka-preferred-replica-election.sh` to ensure preferred replicas are elected before removal.
4	Monitor replica reassignment	Use `kafka-replica-election.sh` to check the status of ongoing reassignments.
5	Shutdown broker	Stop the Kafka process on the target broker after all partitions are reassigned.

Example: Decommissioning a Broker

kafka-configs.sh --bootstrap-server localhost:9092 --entity-type brokers --entity-name 1 --alter --add-config broker.rack=us-west-1

Note: Ensure that the unclean.leader.election.enable is set to false before decommissioning to avoid data loss during leadership changes.

Warning: Removing a broker without proper reassignment can lead to data unavailability or loss. Always verify that all partitions are reassigned before shutting down the broker.

3. Expanding Cluster

Example: Adding a New Broker to Kafka Cluster

# Step 1: Add new broker configuration to server.properties
echo "broker.id=3" >> /opt/kafka/config/server3.properties
echo "listeners=PLAINTEXT://localhost:9093" >> /opt/kafka/config/server3.properties
echo "advertised.listeners=PLAINTEXT://localhost:9093" >> /opt/kafka/config/server3.properties
echo "log.dirs=/tmp/kafka-logs-3" >> /opt/kafka/config/server3.properties

# Step 2: Start the new broker
bin/kafka-server-start.sh config/server3.properties &

# Step 3: Verify the new broker is added to the cluster
bin/kafka-broker-list.sh --bootstrap-server localhost:9092

Configuration Property	Description	Default Value	Notes
broker.id	Unique identifier for the broker in the cluster	-1	Mandatory when starting a new broker
listeners	Network listeners for the broker to accept connections	PLAINTEXT://localhost:9092	Must include a unique port for each broker
advertised.listeners	Address that the broker advertises to clients	Same as listeners	Must be accessible from client machines
log.dirs	Directory where log segments are stored	/tmp/kafka-logs	Must be unique per broker to avoid data corruption

Note: When expanding a Kafka cluster, ensure that the new broker's broker.id is unique and that all brokers can communicate over the network.

Warning: Adding a new broker does not automatically rebalance partitions. Use kafka-reassign-partitions.sh to redistribute data if necessary.

Steps to Expand Kafka Cluster

Create a new server.properties file with unique configuration values
Start the new broker using the updated configuration
Verify that the new broker is recognized by the cluster
Rebalance partitions if needed to distribute load evenly

4. Balancing Partition Load

Property	Default Value	Description
partition.assignment.strategy	`org.apache.kafka.clients.consumer.RangeAssignor`	Strategy used to assign partitions to consumers in a consumer group
replication.factor	1	Number of replicas for each partition
min.insync.replicas	1	Minimum number of replicas that must acknowledge a write for the write to be considered successful

Example: Configuring Partition Assignment Strategy

{
  "partition.assignment.strategy": "org.apache.kafka.clients.consumer.RoundRobinAssignor"
}

Note: The partition.assignment.strategy can be changed to org.apache.kafka.clients.consumer.StickyAssignor for better load distribution in dynamic consumer groups.

Example: Setting Replication Factor

kafka-topics.sh --create --topic my-topic --partitions 6 --replication-factor 3

Warning: Increasing replication.factor beyond the number of available brokers will cause partitions to be unassignable.

Load Balancing Metric	Description	Recommended Value
partition.count	Number of partitions per topic	Even distribution across brokers
leader.count	Number of leaders per broker	As balanced as possible
replica.count	Number of replicas per broker	As balanced as possible

Partition Load Balancing Best Practices

Ensure even distribution of partitions across brokers
Use RoundRobinAssignor or StickyAssignor for better consumer load distribution
Monitor leader and replica counts to detect imbalances
Adjust replication.factor based on fault tolerance needs

5. Rolling Restart

Note: A rolling restart is a maintenance operation that updates brokers one at a time to minimize downtime and ensure cluster availability.

Parameter	Description	Default Value
kafka.server.Running	Indicates whether the Kafka server is running	false
kafka.server.Started	Indicates whether the Kafka server has started	false

Example: Rolling Restart Script

#!/bin/bash
# Stop one broker at a time and restart it
for broker in {0..2}; do
  echo "Stopping broker $broker"
  bin/kafka-server-stop.sh
  sleep 5
  echo "Starting broker $broker"
  bin/kafka-server-start.sh config/server.properties &amp; 
done

Steps to Perform a Rolling Restart

Identify the list of brokers in your Kafka cluster
Stop one broker at a time using kafka-server-stop.sh
Wait for the broker to fully shut down before proceeding
Start the same broker again using kafka-server-start.sh
Repeat for all brokers in the cluster

Best Practices for Rolling Restart

Ensure that the cluster has sufficient replication to handle broker outages
Monitor consumer lag and topic availability during the restart process
Use a script to automate the rolling restart to reduce manual errors
Verify that all brokers are running after the restart

6. Upgrading Kafka Version

Note: Ensure all brokers are running the same version before starting the upgrade process.

Version	Upgrade Path	Notes
Kafka 3.5.x	3.6.x → 3.7.x → 4.0.x	Recommended for production environments
Kafka 3.4.x	3.5.x → 3.6.x → 3.7.x → 4.0.x	Supports rolling upgrades
Kafka 3.3.x	3.4.x → 3.5.x → 3.6.x → 3.7.x → 4.0.x	Requires careful planning and testing

Example: Rolling Upgrade Process

# Step 1: Upgrade one broker at a time
bin/kafka-server-stop.sh
# Update config files and binaries
bin/kafka-server-start.sh

# Step 2: Verify the upgrade
bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic test-topic

Upgrade Workflow

Backup all configurations and data
Test upgrade in a staging environment
Start rolling upgrades from one broker to another
Monitor cluster health and performance
Complete full cluster upgrade once all brokers are updated

Upgrade Considerations

Ensure all clients are compatible with the new Kafka version
Review release notes for breaking changes or deprecated features
Use confluent-hub for managing plugins and components
Plan for downtime if a non-rolling upgrade is required

7. Migrating to KRaft Mode

Kafka's transition from ZooKeeper-based to KRaft (Kafka Raft Metadata) mode represents a significant shift in how the cluster manages metadata. KRaft mode eliminates the dependency on ZooKeeper, offering a more streamlined and scalable approach to cluster management.

Aspect	ZooKeeper Mode	KRaft Mode
Metadata Storage	ZooKeeper	Internal Raft Cluster
Configuration	zookeeper.connect	metadata.log.dir, controller.quorum.voters
Startup Process	Requires ZooKeeper ensemble	Bootstraps internal metadata controller
Cluster Identity	Managed via ZooKeeper	Managed by KRaft controller

Example: KRaft Controller Configuration

# In server.properties for KRaft mode
process.roles=controller
controller.quorum.voters=1001@localhost:9093,1002@localhost:9094
metadata.log.dirs=/tmp/kafka-logs

Example: ZooKeeper Mode Configuration (Legacy)

# In server.properties for ZooKeeper mode
zookeeper.connect=localhost:2181

Note: Migrating to KRaft mode requires careful planning and testing. Ensure all brokers are upgraded to the same version before initiating migration.

Migration Steps Overview

Prepare all brokers with KRaft-compatible versions (2.8+)
Configure process.roles to controller on each broker
Set up controller.quorum.voters with all controller nodes
Start brokers in KRaft mode
Verify cluster health and metadata consistency
Gradually switch clients to KRaft mode

Warning: Once migration to KRaft is initiated, it's not easily reversible. Ensure backups and have rollback plans in place.

8. Handling Broker Failures

Scenario	Description	Action Items
Broker Unavailable	Broker stops responding to client requests or becomes unreachable.	Check network connectivity between brokers and clients. Verify broker logs for errors or warnings. Restart the affected broker if it's unresponsive.
Broker Crash	Broker process crashes unexpectedly, leading to data loss or inconsistency.	Review system logs for crash details (e.g., OOM errors). Ensure Kafka is configured with appropriate `log.flush.interval.messages` and `log.flush.interval.ms`. Recover from backups or replicas if necessary.
Disk Failure	Broker disk becomes full or fails, causing data write issues.	Monitor disk usage via `kafka-log-dirs` tool or monitoring systems. Free up space by deleting old logs or adding more disks. Rebalance partitions to move data off failed disks.

Example: Monitoring Broker Health

kafka-topics.sh --bootstrap-server localhost:9092 --describe --topic my-topic

Note: Use kafka-topics.sh to inspect topic and partition status after broker failures.

Warning: Avoid manual partition reassignment during high load or ongoing consumer processing.

Recovery Steps Summary

Identify the failed broker using kafka-broker-list.sh.
Ensure replication factor is sufficient to maintain availability.
Use kafka-replica-manager.sh to trigger leader election if needed.

Broker Failure vs. Partition Reassignment

Broker Failure: Affects all partitions hosted on that broker.
Partition Reassignment: Moves specific partitions to different brokers without full broker downtime.

9. Managing Disk Space

Example: Configuring Log Retention

{"log.retention.hours": 168,
"log.retention.bytes": -1,
"log.cleanup.policy": "delete"}

Property	Default Value	Description
log.retention.hours	`168`	Number of hours to keep logs before deletion (default 7 days)
log.retention.bytes	`-1`	Maximum size of logs before deletion (-1 = unlimited)
log.cleanup.policy	`delete`	Policy for cleaning up logs (delete or compact)

Note: When log.retention.bytes is set to a positive value, Kafka will delete logs based on size once the limit is exceeded.

Example: Setting Up Disk Space Monitoring

df -h /kafka-data

Comparison: Log Retention Policies

Delete: Removes log segments after retention time or size
Compact: Retains the latest value for each key, useful for stateful systems

10. Configuring Rack Awareness

Rack awareness in Kafka allows you to organize brokers into logical racks (e.g., availability zones or data centers) to improve fault tolerance and replication distribution.

Property	Default Value	Description
broker.rack	`null`	Defines the rack identifier for this broker. Must be set per broker.
client.rack	`null`	Defines the rack identifier for the client. Used by consumers and producers.
replica.lag.time.max.ms	`30000`	Maximum time a replica can lag before being considered out of sync.

Example: Setting up Rack Awareness

# Set broker.rack in server.properties
broker.rack=us-west-1a

# Set client.rack in consumer/producer configs
client.rack=us-west-1a

Note: Rack awareness is used by Kafka's replica.placement.policy to ensure replicas are distributed across different racks for fault tolerance.

Key Points

Rack awareness helps in distributing replicas across different racks to avoid data loss during rack-level failures.
Each broker must have a unique broker.rack value.
The client.rack property is used by clients to identify their rack for optimal partition assignment.