Anyone familiar with Kafka Streams knows how annoying it is to get a “poison pill” message: a bad record that throws an exception and sends your stream into a restart loop. If a stream doesn’t handle these messages properly, it will block the processing of messages from the partition containing the poison pill. This leads to a situation where more time is spent rebalancing the stream than actually running it.

‍

In this post, I want to cover a simple pattern to mitigate this issue: the Dead Letter Topic (DLT). This involves handling poison pills by moving them to a dedicated DLT, allowing the consumer to skip the problematic message and continue processing others.

‍

Is using a DLT enough though?

In many scenarios, using a Dead Letter Topic will suffice. But consider a situation where the order of processing is crucial.

‍

‍

For example, imagine a message at offset 0 that contains a UserLastNameUpdated event. If this message fails to process and is sent to the DLT, the stream won’t be blocked, which is good.

However, when a later message at offset 10 (containing a UserFirstNameUpdated event for the same user) would be processed successfully, it would result in a user with inconsistent information downstream. So, how can we resolve this?

‍

The solution: skip processing of DLT keys

The key is to check if a message’s key is already present on the DLT before processing it from the input topic. If it is, we should skip processing and send it directly to the DLT. This ensures that even if DLT messages are reprocessed later (through a retry mechanism or manual intervention), the original order of events is preserved.

‍

‍

We can achieve this by treating the DLT topic as a GlobalKTable and joining it with our input topic. If a key has no corresponding value in the GlobalKTable, it means that the key is not present on the DLT, and we can continue processing as normal. Otherwise, we will forward the message to the DLT without processing it.

‍

Here’s the relevant code:

record DltJoin(String message, String dlt) {
 
   boolean isOnDlt() {
       return dlt != null;
   }
 
}

var dltTable = builder.globalTable(
       "dlt",
       Materialized.<String, String, KeyValueStore<Bytes, byte[]>>as("dlt-store")
           .withKeySerde(Serdes.String())
           .withValueSerde(Serdes.String()));

var dltJoin = builder.stream("input", Consumed.with(Serdes.String(), Serdes.String()))
       .leftJoin(
               dltTable,
               (k, v) -> k,
               DltJoin::new);

dltJoin.filter((k, v) -> v.isOnDlt())
   .mapValues(DltJoin::message)
   .to("dlt", Produced.with(Serdes.String(), Serdes.String()));

dltJoin.filterNot((k, v) -> v.isOnDlt())
   .mapValues(DltJoin::message)
   .process(...)
   .to("output", Produced.with(Serdes.String(), Serdes.String()));

‍

Important: Because we’re using a GlobalKTable, remember to tombstone successfully reprocessed keys to remove them from the table. This signals to the stream that those keys can be processed normally again.

‍

Simple but elegant

While there are many error handling patterns for KStreams, this DLT and GlobalKTable combination is particularly effective for projects where the order of messages is very important. It’s relatively simple to implement and only requires creating one additional topic (the DLT).

‍

I was inspired to write about due to Kris Van Vlaenderen’s blog on the Sink Gateway. This pattern is used for handling errors when sending data to external systems, so consider exploring the Sink Gateway as well if you need to send data to third-party systems.

‍

If you know of other ways to deal with these poison pills, I’d love to hear about them. Leave a comment below and let me know what you think!