Introduction to Kafka Topics

Kafka topics are fundamental components of Apache Kafka, serving as channels for data communication between producers and consumers. This guide provides an overview of Kafka topics, their characteristics, and how to work with them using Java.

1. What is a Kafka Topic?

A Kafka topic is a logical channel to which messages are sent by producers and from which messages are read by consumers. Topics in Kafka are used to categorize and organize data, allowing multiple producers and consumers to interact with the data streams independently.

2. Key Characteristics of Kafka Topics

• Partitions: Each topic is divided into multiple partitions. Partitions allow Kafka to parallelize processing and increase throughput. Each partition is an ordered, immutable sequence of messages.
• Offset: Each message in a partition is assigned a unique offset, which is a sequential ID that allows consumers to track their progress.
• Replication: Topics can be replicated across multiple brokers to ensure fault tolerance. Each partition has one leader and multiple followers that replicate the data.
• Retention Policy: Kafka topics have configurable retention policies to determine how long messages are kept. Messages can be retained based on time or size.

3. Creating a Kafka Topic

Topics can be created using Kafka's command-line tools. Below is an example of how to create a topic using the `kafka-topics.sh` script.

# Create a new Kafka topic with 3 partitions and a replication factor of 1
bin/kafka-topics.sh --create --topic my-topic --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1
    

4. Listing Kafka Topics

You can list all existing topics in a Kafka cluster using the following command:

# List all Kafka topics
bin/kafka-topics.sh --list --bootstrap-server localhost:9092
    

5. Java Example: Producing to and Consuming from a Kafka Topic

Here's a simple Java example demonstrating how to produce messages to and consume messages from a Kafka topic using the Kafka client library.

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;
import java.util.Collections;

public class KafkaExample {

    public static void main(String[] args) {
        // Producer Configuration
        Properties producerProps = new Properties();
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        producerProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        producerProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // Create Kafka Producer
        KafkaProducer producer = new KafkaProducer<>(producerProps);
        ProducerRecord record = new ProducerRecord<>("my-topic", "key", "value");

        // Send message to topic
        producer.send(record);
        producer.close();

        // Consumer Configuration
        Properties consumerProps = new Properties();
        consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
        consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        // Create Kafka Consumer
        KafkaConsumer consumer = new KafkaConsumer<>(consumerProps);
        consumer.subscribe(Collections.singletonList("my-topic"));

        // Poll for messages
        ConsumerRecords records = consumer.poll(1000);
        for (ConsumerRecord consumerRecord : records) {
            System.out.printf("Offset = %d, Key = %s, Value = %s%n",
                              consumerRecord.offset(), consumerRecord.key(), consumerRecord.value());
        }

        consumer.close();
    }
}
    

6. Kafka Topic Diagram

The following diagram illustrates how topics fit into the Kafka architecture, showing the relationship between producers, topics, partitions, and consumers.

Diagram: Kafka Topics in the Kafka Architecture

7. Conclusion

Kafka topics are essential for organizing and managing data streams within a Kafka cluster. Understanding topics, partitions, and how to interact with them through producers and consumers is crucial for effective Kafka usage.