Operations and Administration

cleanup.policy sets how old data is removed. delete drops whole log segments once they exceed retention.ms (age) or retention.bytes (size). compact keeps the latest value per key forever. retention.ms/bytes only apply to delete.

A new broker joins the cluster but Kafka never auto-moves existing partitions onto it. You must explicitly reassign replicas to the new broker using kafka-reassign-partitions to scale out.

Storage = ingest rate (bytes/sec) x retention seconds x replication factor. So a 10 MB/s topic kept for 7 days at RF=3 needs roughly 10MB x 604800s x 3 = about 18 TB of disk across the cluster.

Use kafka-configs.sh with --bootstrap-server, --entity-type topics, --entity-name <topic>, and --alter --add-config key=value. To remove an override use --delete-config key. --describe shows current overrides.

Static configs live in server.properties and only take effect after a broker restart. Dynamic configs are set with kafka-configs at runtime and apply without restarting the broker, so you can tune a live cluster.

Offline partitions mean some partitions have no leader, so reads/writes fail — page immediately (critical). Under-replicated partitions mean replicas are behind the leader, reducing durability but still serving traffic — warn, page if sustained.

kafka-reassign-partitions is the CLI tool that moves partition replicas between brokers. Its three phases are generate (propose a plan), execute (apply it), and verify (check progress/completion).

A rolling upgrade restarts brokers one at a time, swapping in the new version on each, so the cluster keeps serving while every broker is replaced. One broker is down at a time; the rest keep handling reads and writes.

Turn it on cluster-wide with remote.log.storage.system.enable=true on each broker (plus configure a RemoteStorageManager plugin), then per topic set remote.storage.enable=true. After that, old segments are copied to the remote store.

Run kafka-consumer-groups.sh --bootstrap-server host --describe --group g. It shows per-partition CURRENT-OFFSET, LOG-END-OFFSET, and LAG (end minus current), plus which consumer/member owns each partition. High LAG means consumers are falling behind.

RPO (Recovery Point Objective) = how much data you can lose, measured in time. RTO (Recovery Time Objective) = how long recovery takes. Async cross-cluster replication (MirrorMaker 2) means RPO > 0: anything not yet replicated when the primary dies is lost.

Move every replica off that broker.id to other brokers using kafka-reassign-partitions, wait until it hosts zero partitions, then shut it down and deregister it. Never just kill the process while it holds replicas.

Raw lag (offsets behind) is misleading because acceptable lag depends on throughput. Alert on time-to-drain (lag / consume rate) or sustained-and-growing lag, not a single absolute number, and require the condition to persist before paging.

retention.ms/bytes is total retention across local + remote storage. local.retention.ms/bytes controls how long/how much data stays on local broker disk before it can be deleted (after upload). Local must be <= total, and local sizes your disk.

It rewrites a group's committed offsets so it reprocesses or skips records. Modes include --to-earliest (start of log), --to-latest, --shift-by N (relative), and --to-datetime (offsets at a time). The group must be inactive, and by default it's a dry run unless you add --execute.

Source and target clusters have different offsets for the same record, so you can't reuse raw offsets after failover. MM2's MirrorCheckpointConnector tracks the mapping and writes checkpoints; consumers use RemoteClusterUtils (or automatic sync to __consumer_offsets) to resume at the equivalent position on the target.

Unclean leader election lets Kafka elect an out-of-sync (lagging) replica as leader when no in-sync replica is available. It restores availability but can permanently lose committed messages. It defaults to false.

Each partition replica is open files plus memory and adds controller/recovery work. A rough guide is up to ~1000-4000 partition-replicas per broker. Too many slows leader election, lengthens unclean-shutdown recovery, and increases end-to-end latency.

Throttles cap how fast replicas copy data during a reassignment so it doesn't starve normal traffic. leader.replication.throttled.rate and follower.replication.throttled.rate (bytes/sec, per broker) set the limit; throttled-replicas lists which replicas are throttled.

Phase 1: roll new binaries while pinning inter.broker.protocol.version (and log.message.format.version) to the OLD version, so new and old brokers still speak the old protocol. Phase 2: after all brokers are upgraded, roll again removing/raising the pins. Splitting it keeps a mixed-version cluster compatible and makes phase 1 downgradable.

Set retention.ms=1209600000 (14 days) and ensure retention.bytes is large enough (or -1) so size doesn't evict early. Keep cleanup.policy=delete. Watch disk: 14 days of high throughput can be huge, so plan capacity and consider tiered storage.

On shutdown, the broker asks the controller to move its partition leaderships to other in-sync replicas before it exits, so leader failover is graceful instead of a sudden election. It's on by default.

Partitions = max(target / per-partition producer rate, target / per-partition consumer rate). Each partition is processed by at most one consumer in a group, so partition count caps consumer parallelism. Round up and leave growth headroom.

Use kafka-configs --describe with --entity-type and --entity-name. By default it shows only explicitly-set overrides. Adding --all also lists inherited defaults and each value's source, giving the full effective config.

--entity-name <brokerId> sets a per-broker config that applies to just that broker. --entity-default sets a cluster-wide default applied to all brokers. Per-broker overrides the cluster default for that broker.

The first replica in a partition's replica list is its 'preferred leader'. After failures, leadership drifts off preferred replicas, concentrating load on a few brokers. auto.leader.rebalance.enable=true periodically moves leadership back to preferred replicas when imbalance exceeds a threshold.

Each Kafka API (request type) is versioned. On connect, the client asks the broker which versions it supports (ApiVersions) and uses the highest both understand. Since ~0.10.2 this works both ways: newer clients talk to older brokers and older clients talk to newer brokers. Upgrade brokers and clients independently.

Enable controlled shutdown so a broker moves leadership off itself before stopping. Between brokers, wait until UnderReplicatedPartitions is 0 and there are no offline partitions. Use replication factor 3 with min.insync.replicas 2 and acks=all so producers survive one broker being down.

AdminClient (Admin.create) is the Java API behind the CLI tools. You call methods like createTopics, describeTopics, alterConfigs, createAcls, listConsumerGroups. Each returns a *Result holding KafkaFutures you complete to get values or catch exceptions.