The Project | About Us | Contribute | Donations | License

HOME

In this section, we will delve into the core concepts of messages and offsets in Kafka. Understanding these concepts is crucial for effectively producing and consuming data in Kafka.

What is a Message?

A message in Kafka is a unit of data. It is essentially a key-value pair that is written to and read from Kafka topics. Here are the key components of a Kafka message:

Key: An optional identifier for the message. It can be used to ensure that messages with the same key are sent to the same partition.
Value: The actual data payload of the message.
Timestamp: The time at which the message was produced.
Headers: Optional metadata associated with the message.

Example of a Kafka Message

{
  "key": "user123",
  "value": "User data for user123",
  "timestamp": 1633024800000,
  "headers": {
    "source": "web",
    "type": "user-update"
  }
}

What is an Offset?

An offset is a unique identifier assigned to each message within a partition. It is a sequential number that Kafka uses to keep track of the order of messages. Offsets are crucial for ensuring that messages are consumed in the correct order and for enabling consumers to resume reading from a specific point in the partition.

Key Points about Offsets

Sequential: Offsets are assigned in a sequential manner within a partition.
Unique: Each message within a partition has a unique offset.
Persistent: Offsets are stored persistently, allowing consumers to resume from the last read offset.

How Offsets Work

When a producer sends a message to a Kafka topic, the message is appended to the end of a partition, and an offset is assigned to it. Consumers read messages from a partition by specifying the offset from which they want to start reading.

Example of Offsets in a Partition

Offset	Key	Value
0	user123	User data for user123
1	user456	User data for user456
2	user789	User data for user789

In this example, the partition contains three messages with offsets 0, 1, and 2.

Practical Example: Producing and Consuming Messages with Offsets

Producing Messages

Let's produce some messages to a Kafka topic named user-updates.

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;

public class ProducerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        for (int i = 0; i < 3; i++) {
            String key = "user" + i;
            String value = "User data for user" + i;
            ProducerRecord<String, String> record = new ProducerRecord<>("user-updates", key, value);
            producer.send(record);
        }

        producer.close();
    }
}

Consuming Messages

Now, let's consume the messages from the user-updates topic.

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Collections;
import java.util.Properties;

public class ConsumerExample {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "user-update-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("user-updates"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("Offset: %d, Key: %s, Value: %s%n", record.offset(), record.key(), record.value());
            }
        }
    }
}

Practical Exercise

Exercise: Produce and Consume Messages

Task: Write a producer that sends 5 messages to a topic named test-topic. Each message should have a unique key and value.
Task: Write a consumer that reads messages from test-topic and prints the offset, key, and value of each message.

Solution

Producer

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;

public class TestProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        for (int i = 0; i < 5; i++) {
            String key = "key" + i;
            String value = "value" + i;
            ProducerRecord<String, String> record = new ProducerRecord<>("test-topic", key, value);
            producer.send(record);
        }

        producer.close();
    }
}

Consumer

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import java.util.Collections;
import java.util.Properties;

public class TestConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "test-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Collections.singletonList("test-topic"));

        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("Offset: %d, Key: %s, Value: %s%n", record.offset(), record.key(), record.value());
            }
        }
    }
}

Common Mistakes and Tips

Mistake: Not closing the producer after sending messages.
- Tip: Always close the producer to ensure all messages are flushed and sent.
Mistake: Not handling exceptions in producer and consumer code.
- Tip: Implement proper exception handling to manage errors gracefully.
Mistake: Using the same group ID for different consumers that should not share offsets.
- Tip: Use unique group IDs for different consumer groups to avoid offset conflicts.

Conclusion

In this section, we covered the fundamental concepts of messages and offsets in Kafka. We learned how messages are structured, how offsets work, and how to produce and consume messages using Java. Understanding these concepts is essential for working effectively with Kafka. In the next section, we will explore Kafka operations in more detail, including producing and consuming messages programmatically.

Messages and Offsets

What is a Message?

Example of a Kafka Message

What is an Offset?

Key Points about Offsets

How Offsets Work

Example of Offsets in a Partition

Practical Example: Producing and Consuming Messages with Offsets

Producing Messages

Consuming Messages

Practical Exercise

Exercise: Produce and Consume Messages

Solution

Producer

Consumer

Common Mistakes and Tips

Conclusion

Kafka Course

Module 1: Introduction to Kafka

Module 2: Kafka Core Concepts

Module 3: Kafka Operations

Module 4: Kafka Configuration and Management

Module 5: Advanced Kafka Topics

Module 6: Kafka Ecosystem and Integrations

Module 7: Kafka Case Studies and Best Practices