Introduction to Kafka

Kafka is a distributed event streaming platform that functions as a high-performance messaging system. It enables building real-time data pipelines, stream processing applications, and event-driven architectures at scale. Unlike traditional message brokers, Kafka combines messaging, storage, and stream processing capabilities in a single horizontally scalable system.

Core Architecture

Key Components

A Kafka cluster consists of multiple brokers that coordinate through a metadata management system (ZooKeeper in versions prior to 2.8.0, KRaft mode in modern deployments). Data is organized into subjects (commonly called topics), which are divided into partitions for parallelism and scalability. Each partition maintains an ordered, immutable sequence of records.

Replication ensures fault tolerance: every partition has a leader replica and multiple follower replicas. The leader handles all read and write operations, while followers continuously replicate the leader's data. The set of in-sync replicas (ISR) includes all followers that have caught up within a configurable time window (default 30 seconds).

Deployment Setup

After extracting the Kafka distribution archive, configure each broker through server.properties:

# Unique broker identifier
broker.id=0

# Log storage directories
log.dirs=/opt/kafka/data

# Connection string for metadata management
# For ZooKeeper: zookeeper.connect=zk1:2181,zk2:2181,zk3:2181/kafka
# For KRaft: controller.quorum.voters=1@controller1:9093,2@controller2:9093

Start brokers using the provided scripts:

bin/kafka-server-start.sh -daemon config/server.properties

Producer Implementation

Message Transmission Flow

The producer client operates two threads: the main application thread and a background sender thread. The main thread writes messages to a RecordAccumulator buffer, which batches records by partition. The sender thread polls this buffer and transmits batches to brokers when size or time thresholds are met.

Basic Producer Configuration

Properties config = new Properties();
config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092");
config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

KafkaProducer<string string=""> producer = new KafkaProducer<>(config);
</string>

Asynchronous Transmission Patterns

Fire-and-Forget

try (KafkaProducer<string string=""> eventProducer = new KafkaProducer<>(config)) {
    for (int counter = 0; counter < 5; counter++) {
        eventProducer.send(new ProducerRecord<>("sample-topic", "payload-" + counter));
    }
}
</string>

Callback-Enhanced Sending

producer.send(new ProducerRecord<>("sample-topic", "record-" + index), (metadata, exception) -> {
    if (exception == null) {
        System.out.printf("Delivery successful - Topic: %s, Partition: %d, Offset: %d%n",
                          metadata.topic(), metadata.partition(), metadata.offset());
    } else {
        exception.printStackTrace();
    }
});

Synchronous Transmission

Convert asynchronous sands to synchronous by invoking get() on the returned Future:

try {
    RecordMetadata metadata = producer.send(record).get();
    System.out.printf("Synchronous send complete: %s%n", metadata);
} catch (Exception e) {
    e.printStackTrace();
}

Custom Partitioning Strategy

Implement the Partitioner interface to control message routing:

public class PriorityPartitioner implements Partitioner {
    @Override
    public int partition(String subject, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
        String payload = value.toString();
        return payload.startsWith("critical-") ? 0 : 1;
    }

    @Override public void close() {}
    @Override public void configure(Map<string> configs) {}
}

// Register the partitioner
config.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, PriorityPartitioner.class.getName());
</string>

Performence Optimization

Increase throughput by adjusting batching and compression:

config.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384); // 16KB batches
config.put(ProducerConfig.LINGER_MS_CONFIG, 5); // Wait up to 5ms
config.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432L); // 32MB buffer
config.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, CompressionType.SNAPPY.name);

Reliability Guarantees

The acks parameter controls durability:

• acks=0: No acknowledgment, potential data loss
• acks=1: Leader acknowledgment, risk of loss during leader failure
• acks=all: ISR acknowledgment, strongest guarantee

config.put(ProducerConfig.ACKS_CONFIG, "all");
config.put(ProducerConfig.RETRIES_CONFIG, 3);

Idempotency and Transactions

Enable exactly-once semantics by combining idempotency with transactions:

config.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
config.put(ProducerConfig.ACKS_CONFIG, "all");
config.put(ProducerConfig.RETRIES_CONFIG, 3);
config.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "unique-txn-id-001");

KafkaProducer<string string=""> transactionalProducer = new KafkaProducer<>(config);
transactionalProducer.initTransactions();

try {
    transactionalProducer.beginTransaction();
    for (int i = 0; i < 10; i++) {
        transactionalProducer.send(new ProducerRecord<>("sample-topic", "txn-msg-" + i));
    }
    transactionalProducer.commitTransaction();
} catch (ProducerFencedException e) {
    transactionalProducer.abortTransaction();
} finally {
    transactionalProducer.close();
}
</string>

Broker Internals

Metadata Management

In ZooKeeper-based deployments, the following metadata is stored:

• Active broker registry
• Partition leadership and ISR information
• Controller election state

Modern KRaft mode stores this metadata within Kafka itself, eliminating external dependencies.

Replication and Leader Election

When a leader fails, the controller selects a new leader from the ISR based on replica priority. Followers that fall behind beyond replica.lag.time.max.ms are removed from ISR to prevent stale data promotion.

Storage Mechanism

Each partition maps to a directory containing segment files: .log stores message data, .index provides sparse offset indexing (one entry per 4KB of data by default). This design enables efficient sequential I/O and fast lookups.

Log Retention

Kafka provides two cleanup policies:

• delete: Remove segments older than retention.ms or larger than retention.bytes
• compact: Retain only the latest message per key within a partition

Consumer Implementation

Pull-Based Consumption Model

Consumers actively fetch records from brokers, allowing them to control consumption rate and avoid overwhelming slow consumers.

Consumer Group Mechanics

Consumers coordinate through group membership:

• Each group receives each partition's data exactly once
• Partition assignments are rebalanced when members join or leave
• Heartbeat mechanism maintains membership (default 3-second interval)

Essential Consumer Configuration

Properties config = new Properties();
config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "broker1:9092,broker2:9092");
config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
config.put(ConsumerConfig.GROUP_ID_CONFIG, "consumer-group-1");
config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // Start from earliest if no offset

KafkaConsumer<string string=""> consumer = new KafkaConsumer<>(config);
consumer.subscribe(Collections.singletonList("sample-topic"));
</string>

Offset Management Strategies

Automatic Commit

config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
config.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 5000);

Manual Commit

config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

while (true) {
    ConsumerRecords<string string=""> records = consumer.poll(Duration.ofSeconds(1));
    records.forEach(record -> processRecord(record));
    consumer.commitSync(); // Synchronous commit
    // or consumer.commitAsync(); // Asynchronous commit
}
</string>

Seeking to Specific Offsets

Set<topicpartition> assignedPartitions = consumer.assignment();
while (assignedPartitions.isEmpty()) {
    consumer.poll(Duration.ofSeconds(1));
    assignedPartitions = consumer.assignment();
}

assignedPartitions.forEach(partition -> consumer.seek(partition, 100L));
</topicpartition>

Time-Based Offset Lookup

Map<topicpartition long=""> partitionTimestampMap = new HashMap<>();
assignedPartitions.forEach(partition -> 
    partitionTimestampMap.put(partition, System.currentTimeMillis() - TimeUnit.DAYS.toMillis(1))
);

Map<topicpartition offsetandtimestamp=""> offsetMap = consumer.offsetsForTimes(partitionTimestampMap);
offsetMap.forEach((partition, offsetAndTs) -> {
    if (offsetAndTs != null) {
        consumer.seek(partition, offsetAndTs.offset());
    }
});
</topicpartition></topicpartition>

Partition Assignment Policies

Kafka supports pluggable assignment strategies:

• RangeAssignor: Allocates contiguous partition ranges per topic (can cause imbalance)
• RoundRobinAssignor: Distributes partitions across all consumers uniformly
• StickyAssignor: Balances load while minimizing partition movement during rebalancing

config.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY_CONFIG, 
           RoundRobinAssignor.class.getName());

Performance Tuning

Address consumer lag by:

• Increasing fetch.min.bytes and fetch.max.wait.ms for larger batches
• Raising max.poll.records to process more per iteration
• Scaling consumer instances within the group
• Adding partitions to increase parallelism

Operational Monitoring

Kafka Eagle provides comprehensive cluster monitoring through a web dashboard. It tracks broker health, topic metrics, consumer lag, and offset positions. Deployment requires a MySQL backend for storing historical metrics.

Modern KRaft Mode

Since version 2.8.0, Kafka can operate without ZooKeeper using the KRaft consensus protocol. The controller nodes manage metadata internally, simplifying deployment and improving scalability. Configure KRaft through process.roles and controller.quorum.voters settings.