Implementing Kafka and Dead Letter Queue in Java

In this article, I will share my knowledge and experience integrating Kafka with a Java Spring Boot application. Firstly, we will discuss Kafka, and then we will delve into the integration process.

What is Kafka

Kafka is used to build real-time streaming data pipelines and real-time streaming applications. A data pipeline reliably processes and moves data from one system to another, and a streaming application is an application that consumes streams of data. For example, if you want to create a data pipeline that takes in user activity data to track how people use your website in real-time, Kafka would be used to ingest and store streaming data. At the same time, serving reads for the applications powering the data pipeline. Kafka is also often used as a message broker solution, which is a platform that processes and mediates communication between two applications.
Kafka was originally developed by LinkedIn and was subsequently open-sourced in early 2011. Kafka was co-created by Jay Kreps, Neha Narkhede, and Jun Rao. Jay Kreps named the software after the author Franz Kafka because it is “a system optimized for writing,” and he admired Kafka’s work.

Terminology

• Message: In Kafka, data is often considered a set of messages. A message is a simple array of bytes.
• Producer: An application that sends messages. It does not send messages directly to the recipient but sends them to the Kafka server.
• Consumer: An application that reads messages from the Kafka server.
• Kafka Broker: A Kafka server that acts as a message broker between the producer and consumer since they do not connect directly.
• Kafka Cluster: Kafka operates as a cluster — a group of computers sharing workload for a common purpose. Each instance contains a Kafka broker.
• Topic: A category or feed name to which messages are published. Producers send messages to topics on Kafka brokers, and consumers read messages from these topics.
• Partitions: Each topic is divided into partitions. The number of partitions is configured when creating the topic using the partitions config. Kafka distributes these partitions among the brokers in the cluster with the help of ZooKeeper. If the number of partitions exceeds the number of brokers, a broker can have multiple partitions of the same topic assigned to it.
• Partition Replicas: Each partition has one leader broker and n follower brokers. The number of replica brokers is configured using the “replication.factor” config when creating a topic. For example, setting “replication.factor” to 3 means the partition is replicated on three brokers: one leader and two followers. Producers write only to the leader broker, and followers asynchronously replicate the data. The replication factor cannot exceed the total number of brokers in the cluster; otherwise, Kafka throws an InvalidReplicationFactorException. This ensures that each replica is on a separate broker to prevent data loss if a broker fails. A replication factor of 1 means no replication and is typically used for development purposes.
• Offsets: Once an offset is assigned to a message, it never changes. The first message in a partition has an offset of zero, the next one has an offset of one, and so on. Offsets are local to the partition. To locate a message, you need the topic name, partition number, and offset.
• Acknowledgements: The acks config is a parameter set on a producer that denotes the number of brokers that must acknowledge receipt of a message before considering the write successful. It supports three values:
• acks=0: The producer does not wait for any acknowledgement from the broker. This offers the highest throughput but can lead to lost messages — a fire-and-forget strategy.
• acks=1: The leader broker sends an acknowledgement immediately after receiving the message. Followers replicate the data asynchronously. This provides eventual consistency.
• acks=all: The leader broker waits for all replicas to acknowledge receipt before sending an acknowledgement to the producer. This ensures strong consistency but with higher latency and lower throughput.

Default acks Values:

• For Kafka versions before 3.0, the default is acks=1.
• For Kafka versions 3.0 and later, the default is acks=all.

Producer

A producer is an application that sends data into topics. Here is our producer code :

@Service
@RequiredArgsConstructor
public class VaccineReminderProducer {

    private final KafkaTemplate<String, String> kafkaTemplate;

    private final ObjectMapper objectMapper;

    private final KafkaConfig kafkaConfig;

    public void sendVaccineReminder(VaccineReminder reminder) throws JsonProcessingException {
        String message = objectMapper.writeValueAsString(reminder);
        kafkaTemplate.send(kafkaConfig.getVaccineReminderTopic(), message);
    }
}

The producer is responsible for sending messages to the topic. When the sendVaccineReminder method is called, the VaccineReminder object is converted to a string. This string, referred to as message, is then sent to the specified Kafka topic.

Here is the VaccineReminder class:

@Getter
@Setter
@AllArgsConstructor
@NoArgsConstructor
@Builder
public class VaccineReminder {
    private String chatId;
    private String message;
    private String userId;
}

Consumer

A consumer is an application that subscribes to events. Here is our consumer code :

@Service
@RequiredArgsConstructor
public class VaccineReminderConsumer {

    private final VaccineReminderBot vaccineReminderBot;
    private final KafkaTemplate<String, String> kafkaTemplate;
    private final KafkaConfig kafkaConfig;
    private final ObjectMapper objectMapper;

    @KafkaListener(topics = "${kafka.topic.reminder-topic}", groupId = "${spring.kafka.consumer.group-id}")
    public void handleVaccineReminder(@Payload String message) {
        try {
            VaccineReminder reminder = objectMapper.readValue(message, VaccineReminder.class);
            vaccineReminderBot.sendMessage(reminder.getChatId(), reminder.getMessage());
        } catch (Exception e) {
            sendToDeadLetterQueue(String.format("Cause : %s, failed message : %s", e.getMessage(), message));
        }
    }

    private void sendToDeadLetterQueue(String message) {
        kafkaTemplate.send(kafkaConfig.getVaccineReminderDeadLetterTopic(), message);
    }
}

The consumer reads and processes data from the topic. When a new message arrives, it attempts to convert it into a VaccineReminder object and sends the message via the bot. If an error occurs during this process, the message is sent to the dead-letter queue.

Dead Letter Queue (DLQ)

Effectively identifying and managing errors is vital for any dependable data streaming pipeline. A Dead Letter Queue (DLQ) acts as a specialized component within a messaging system or data streaming platform, designed to store messages that haven’t been processed successfully. Instead of simply discarding these problematic messages, the system redirects them to the DLQ for further analysis and handling. In systems like Kafka, messages often end up in a DLQ due to issues like incorrect message formatting or missing and invalid content.

Here is our DLQ consumer code :

@Service
@RequiredArgsConstructor
public class DeadLetterQueueConsumer {

    private static final Logger logger = LoggerFactory.getLogger(DeadLetterQueueConsumer.class);

    @KafkaListener(topics = "vaccine-dead-letter-topic", groupId = "vaccine-dlq-group")
    public void handleDeadLetter(VaccineReminder reminder) {
        logger.error("Dead-letter message: {}", reminder);
    }
}

This simple DLQ consumer class logs every failed message. You can customize this part according to your needs, such as creating metrics for failed messages, sending them for manual inspection, or notifying via a bot.

DLQ Error Handling Strategies

Several options are available for handling messages stored in a dead letter queue:

1. Re-process: Some messages in the DLQ need to be re-processed after fixing the underlying issue. This solution might involve an automatic script, human intervention to edit the message, or returning an error to the producer to resend the corrected message.
2. Drop Bad Messages (After Analysis): Depending on your setup, bad messages might be expected. Before dropping them, a business process should examine these messages. For instance, a dashboard application can consume and visualize the error messages.
3. Advanced Analytics: Instead of processing each message individually, analyze the incoming data for real-time insights or issues. A simple ksqlDB application can perform stream processing for calculations, such as the average number of error messages per hour, helping you make informed decisions about errors in your Kafka applications.
4. Stop the Workflow: If bad messages are rare, you might choose to stop the overall business process. This action can be automated or decided by a human. The DLQ externalizes the problem and decision-making, allowing you to address the issue appropriately.
5. Ignore: In some use cases, it might be acceptable to let the DLQ fill up without immediate action. This approach can be useful for monitoring the overall behaviour of the Kafka application. Remember that a Kafka topic has a retention time, and messages are removed after that period. Set this up correctly for your needs and monitor the DLQ for unexpected behaviour, such as rapid growth.

You can find my example project here. It includes various technologies like Keycloak and Swagger. Stay tuned for upcoming articles.

Thank you for reading, and happy coding!