Handling bad messages using Kafka's Streams API

Question

I have a basic stream processing flow which looks like

master topic -> my processing in a mapper/filter -> output topics

and I am wondering about the best way to handle "bad messages". This could potentially be things like messages that I can't deserialize properly, or perhaps the processing/filtering logic fails in some unexpected way (I have no external dependencies so there should be no transient errors of that sort).

I was considering wrapping all my processing/filtering code in a try catch and if an exception was raised then routing to an "error topic". Then I can study the message and modify it or fix my code as appropriate and then replay it on to master. If I let any exceptions propagate, the stream seems to get jammed and no more messages are picked up.

• Is this approach considered best practice?
• Is there a convenient Kafka streams way to handle this? I don't think there is a concept of a DLQ...
• What are the alternative ways to stop Kafka jamming on a "bad message"?
• What alternative error handling approaches are there?

For completeness here is my code (pseudo-ish):

class Document {
    // Fields
}

class AnalysedDocument {

    Document document;
    String rawValue;
    Exception exception;
    Analysis analysis;

    // All being well
    AnalysedDocument(Document document, Analysis analysis) {...}

    // Analysis failed
    AnalysedDocument(Document document, Exception exception) {...}

    // Deserialisation failed
    AnalysedDocument(String rawValue, Exception exception) {...}
}

KStreamBuilder builder = new KStreamBuilder();
KStream<String, AnalysedPolecatDocument> analysedDocumentStream = builder
    .stream(Serdes.String(), Serdes.String(), "master")
    .mapValues(new ValueMapper<String, AnalysedDocument>() {
         @Override
         public AnalysedDocument apply(String rawValue) {
             Document document;
             try {
                 // Deserialise
                 document = ...
             } catch (Exception e) {
                 return new AnalysedDocument(rawValue, exception);
             }
             try {
                 // Perform analysis
                 Analysis analysis = ...
                 return new AnalysedDocument(document, analysis);
             } catch (Exception e) {
                 return new AnalysedDocument(document, exception);
             }
         }
    });

// Branch based on whether analysis mapping failed to produce errorStream and successStream
errorStream.to(Serdes.String(), customPojoSerde(), "error");
successStream.to(Serdes.String(), customPojoSerde(), "analysed");

KafkaStreams streams = new KafkaStreams(builder, config);
streams.start();

Any help greatly appreciated.

Ponton

1> The quarantine topic approach seems risky as a bad producer could result in high overhead, especially if multiple consumers of that topic keep busy pushing the same malformed message to that quarantine topic 2> The flatMap approach sounds more intuitive, and potential re-partitioning overhead could be minimized with KStream<byte[], Long> doubled = input.flatMap( .. validate deserialization of k and v and have the drawback of having to deserialize (safely this time) the key again; as the cost (of deserialization) of the key is much less than the cost for the value – Tehee

Answer 1

Right now, Kafka Streams offers only limited error handling capabilities. There is work in progress to simplify this. For now, your overall approach seems to be a good way to go.

One comment about handling de/serialization errors: handling those error manually, requires you to do de/serialization "manually". This means, you need to configure ByteArraySerdes for key and value for you input/output topic of your Streams app and add a map() that does the de/serialization (ie, KStream<byte[],byte[]> -> map() -> KStream<keyType,valueType> -- or the other way round if you also want to catch serialization exceptions). Otherwise, you cannot try-catch deserialization exceptions.

With your current approach, you "only" validate that the given string represents a valid document -- but it could be the case, that the message itself is corrupted and cannot be converted into a String in the source operator in the first place. Thus, you don't actually cover deserialization exception with you code. However, if you are sure a deserialization exception can never happen, you approach would be sufficient, too.

Update

This issues is tackled via KIP-161 and will be included in the next release 1.0.0. It allows you to register an callback via parameter default.deserialization.exception.handler. The handler will be invoked every time a exception occurs during deserialization and allows you to return an DeserializationResponse (CONTINUE -> drop the record an move on, or FAIL that is the default).

Update 2

With KIP-210 (will be part of in Kafka 1.1) it's also possible to handle errors on the producer side, similar to the consumer part, by registering a ProductionExceptionHandler via config default.production.exception.handler that can return CONTINUE.

Gunner

An excellent answer as always Matthias. Nice to know I am on the right track. I will make the suggested changes. I guess I should also include a "safe mapper" to go to byte array as well (instead of using a custom serde in the "to" statement). KStream<byte[],byte[]> -> map() -> KStream<keyType,valueType> -> KStream<byte[],byte[]> – Ponton

I want to point that ProductionExceptionHandler doesn't cope with the business logic exceptions but only with issues faced by the actual kafka producer. – Languishing

Answer 2

Update Mar 23, 2018: Kafka 1.0 provides much better and easier handling for bad error messages ("poison pills") via KIP-161 than what I described below. See default.deserialization.exception.handler in the Kafka 1.0 docs.

This could potentially be things like messages that I can't deserialize properly [...]

Ok, my answer here focuses on the (de)serialization issues as this might be the most tricky scenario to handle for most users.

[...] or perhaps the processing/filtering logic fails in some unexpected way (I have no external dependencies so there should be no transient errors of that sort).

The same thinking (for deserialization) can also be applied to failures in the processing logic. Here, most people tend to gravitate towards option 2 below (minus the deserialization part), but YMMV.

I was considering wrapping all my processing/filtering code in a try catch and if an exception was raised then routing to an "error topic". Then I can study the message and modify it or fix my code as appropriate and then replay it on to master. If I let any exceptions propagate, the stream seems to get jammed and no more messages are picked up.

• Is this approach considered best practice?

Yes, at the moment this is the way to go. Essentially, the two most common patterns are (1) skipping corrupted messages or (2) sending corrupted records to a quarantine topic aka a dead letter queue.

• Is there a convenient Kafka streams way to handle this? I don't think there is a concept of a DLQ...

Yes, there is a way to handle this, including the use of a dead letter queue. However, it's (at least IMHO) not that convenient yet. If you have any feedback on how the API should allow you to handle this -- e.g. via a new or updated method, a configuration setting ("if serialization/deserialization fails send the problematic record to THIS quarantine topic") -- please let us know. :-)

• What are the alternative ways to stop Kafka jamming on a "bad message"?
• What alternative error handling approaches are there?

See my examples below.

FWIW, the Kafka community is also discussing the addition of a new CLI tool that allows you to skip over corrupted messages. However, as a user of the Kafka Streams API, I think ideally you want to handle such scenarios directly in your code, and fallback to CLI utilities only as a last resort.

Here are some patterns for the Kafka Streams DSL to handle corrupted records/messages aka "poison pills". This is taken from http://docs.confluent.io/current/streams/faq.html#handling-corrupted-records-and-deserialization-errors-poison-pill-messages

Option 1: Skip corrupted records with flatMap

This is arguably what most users would like to do.

• We use flatMap because it allows you to output zero, one, or more output records per input record. In the case of a corrupted record we output nothing (zero records), thereby ignoring/skipping the corrupted record.
• Benefit of this approach compared to the others ones listed here: We need to manually deserialize a record only once!
• Drawback of this approach: flatMap "marks" the input stream for potential data re-partitioning, i.e. if you perform a key-based operation such as groupings (groupBy/groupByKey) or joins afterwards, your data will be re-partitioned behind the scenes. Since this might be a costly step we don't want that to happen unnecessarily. If you KNOW that the record keys are always valid OR that you don't need to operate on the keys (thus keeping them as "raw" keys in byte[] format), you can change from flatMap to flatMapValues, which will not result in data re-partitioning even if you join/group/aggregate the stream later.

Code example:

Serde<byte[]> bytesSerde = Serdes.ByteArray();
Serde<String> stringSerde = Serdes.String();
Serde<Long> longSerde = Serdes.Long();

// Input topic, which might contain corrupted messages
KStream<byte[], byte[]> input = builder.stream(bytesSerde, bytesSerde, inputTopic);

// Note how the returned stream is of type KStream<String, Long>,
// rather than KStream<byte[], byte[]>.
KStream<String, Long> doubled = input.flatMap(
    (k, v) -> {
      try {
        // Attempt deserialization
        String key = stringSerde.deserializer().deserialize(inputTopic, k);
        long value = longSerde.deserializer().deserialize(inputTopic, v);

        // Ok, the record is valid (not corrupted).  Let's take the
        // opportunity to also process the record in some way so that
        // we haven't paid the deserialization cost just for "poison pill"
        // checking.
        return Collections.singletonList(KeyValue.pair(key, 2 * value));
      }
      catch (SerializationException e) {
        // log + ignore/skip the corrupted message
        System.err.println("Could not deserialize record: " + e.getMessage());
      }
      return Collections.emptyList();
    }
);

Option 2: dead letter queue with branch

Compared to option 1 (which ignores corrupted records) option 2 retains corrupted messages by filtering them out of the "main" input stream and writing them to a quarantine topic (think: dead letter queue). The drawback is that, for valid records, we must pay the manual deserialization cost twice.

KStream<byte[], byte[]> input = ...;

KStream<byte[], byte[]>[] partitioned = input.branch(
    (k, v) -> {
      boolean isValidRecord = false;
      try {
        stringSerde.deserializer().deserialize(inputTopic, k);
        longSerde.deserializer().deserialize(inputTopic, v);
        isValidRecord = true;
      }
      catch (SerializationException ignored) {}
      return isValidRecord;
    },
    (k, v) -> true
);

// partitioned[0] is the KStream<byte[], byte[]> that contains
// only valid records.  partitioned[1] contains only corrupted
// records and thus acts as a "dead letter queue".
KStream<String, Long> doubled = partitioned[0].map(
    (key, value) -> KeyValue.pair(
        // Must deserialize a second time unfortunately.
        stringSerde.deserializer().deserialize(inputTopic, key),
        2 * longSerde.deserializer().deserialize(inputTopic, value)));

// Don't forget to actually write the dead letter queue back to Kafka!
partitioned[1].to(Serdes.ByteArray(), Serdes.ByteArray(), "quarantine-topic");

Option 3: Skip corrupted records with filter

I only mention this for completeness. This option looks like a mix of options 1 and 2, but is worse than either of them. Compared to option 1, you must pay the manual deserialization cost for valid records twice (bad!). Compared to option 2, you lose the ability to retain corrupted records in a dead letter queue.

KStream<byte[], byte[]> validRecordsOnly = input.filter(
    (k, v) -> {
      boolean isValidRecord = false;
      try {
        bytesSerde.deserializer().deserialize(inputTopic, k);
        longSerde.deserializer().deserialize(inputTopic, v);
        isValidRecord = true;
      }
      catch (SerializationException e) {
        // log + ignore/skip the corrupted message
        System.err.println("Could not deserialize record: " + e.getMessage());
      }
      return isValidRecord;
    }
);
KStream<String, Long> doubled = validRecordsOnly.map(
    (key, value) -> KeyValue.pair(
        // Must deserialize a second time unfortunately.
        stringSerde.deserializer().deserialize(inputTopic, key),
        2 * longSerde.deserializer().deserialize(inputTopic, value)));

Any help greatly appreciated.

I hope I could help. If yes, I'd appreciate your feedback on how we could improve the Kafka Streams API to handle failures/exceptions in a better/more convenient way than today. :-)

Vasilikivasilis

Hi Miguno, I think in general having the user handle the errors themselves makes sense and provides the flexibility that seems core to the library's philosophy. Some examples or a cookbook somewhere for different solutions might be nice I guess. – Ponton

Please could you explain your comment about re-partitioning regarding option 1? Would this re-partitioning overhead still occur if I was writing the resultant stream ("doubled" in your example) to a different topic? – Ponton

Re-partitioning would only occur if you perform a key-based operation such as groupBy or leftJoin. – Vasilikivasilis

Understood. So not an issue if just routing the filtered stream to somewhere else. Perfect. I think I will use a combination of 1 and 2. Keep up the great work. – Ponton

Oh, to clarify my comment above: If you do use to or through after flatMap then re-partitioning will happen in the to/through because, by definition, to and through are writing data back to Kafka, and that must be done by output topic partitions. – Vasilikivasilis

Just for completeness: there is a JIRA to allow user to skip a repartitioning step if the user knows it's not required: issues.apache.org/jira/browse/KAFKA-4835 – Gunner

There is a KIP open asking for general feedback and comments and you can add yours on the discuss thread to improve exception handling: cwiki.apache.org/confluence/display/KAFKA/…. FYI. – Hayfield

For Option 3 couldn't you avoid the double deserialization by using an envelope object that you map to. The envelope object would contain a boolean indicating the validity of the message, and if valid, the actual deserialized message itself. i think that would then provide a good generic solution. – Obolus

I have the similar problem for KTabke and KGroupTable, can you please share the code snipper for handling error message in that as well – Mettlesome

can you please check this: #63468679 Please let me know if I am missing stomething – Wessling

Answer 3

For the processing logic you could take this approach:

someKStream 

    .mapValues(inputValue -> {
        // for each execution the below "return" could provide a different class than the previous run!
        // e.g. "return isFailedProcessing ? failValue : successValue;" 
        // where failValue and successValue have no related classes
        return someObject; // someObject class vary at runtime depending on your business
    }) // here you'll have KStream<whateverKeyClass, Object> -> yes, Object for the value!

    // you could have a different logic for choosing  
    // the target topic, below is just an example
    .to((k, v, recordContext) -> v instanceof failValueClass ?
            "dead-letter-topic" : "success-topic",
            // you could completelly ignore the "Produced" part 
            // and rely on spring-boot properties only, e.g. 
            // spring.kafka.streams.properties.default.key.serde=yourKeySerde
            // spring.kafka.streams.properties.default.value.serde=org.springframework.kafka.support.serializer.JsonSerde
            Produced.with(yourKeySerde, 
                            // JsonSerde could be an instance configured as you need 
                            // (with type mappings or headers setting disabled, etc)
                            new JsonSerde<>())); 

Your classes, though different and landing into different topics, will serialize as expected.

When not using to(), but instead one wants to continue with other processing, he could use branch() with splitting the logic based on the kafka-value class; the trick for branch() is to return KStream<keyClass, ?>[] in order to further allow one to cast to the appropriate class the individual array items.

Languishing

Answer 4

If you want to send an exception (custom exception) to another topic (ERROR_TOPIC_NAME):

@Bean
  public KStream<String, ?> kafkaStreamInput(StreamsBuilder kStreamBuilder) {
    KStream<String, InputModel> input = kStreamBuilder.stream(INPUT_TOPIC_NAME);
    return service.messageHandler(input);
  }

public KStream<String, ?> messageHandler(KStream<String, InputModel> inputTopic) {
    KStream<String, Object> output;
    output = inputTopic.mapValues(v -> {
      try {
        //return InputModel
        return normalMethod(v);
      } catch (Exception e) {
        //return ErrorModel
        return errorHandler(e);
      }
    });
  
    output.filter((k, v) -> (v instanceof ErrorModel)).to(KafkaStreamsConfig.ERROR_TOPIC_NAME);
    output.filter((k, v) -> (v instanceof InputModel)).to(KafkaStreamsConfig.OUTPUT_TOPIC_NAME);

    return output;
  }

If you want to handle Kafka exceptions and skip it:

@Autowired
  public ConsumerErrorHandler(
      KafkaProducer<String, ErrorModel> dlqProducer) {
    this.dlqProducer = dlqProducer;
  }

  @Bean
  ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
      ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
      ObjectProvider<ConsumerFactory<Object, Object>> kafkaConsumerFactory) {
    ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
    configurer.configure(factory, kafkaConsumerFactory.getIfAvailable());
    factory.setErrorHandler(((exception, data) -> {

      ErrorModel errorModel = ErrorModel.builder().message()
          .status("500").build();
      assert data != null;
        dlqProducer.send(new ProducerRecord<>(DLQ_TOPIC, data.key().toString(), errorModel));
    }));
    return factory;
  }

Iridissa

Your answer could be improved with additional supporting information. Please edit to add further details, such as citations or documentation, so that others can confirm that your answer is correct. You can find more information on how to write good answers in the help center. – Fourscore

Answer 5

All above answers although valid and useful, they are assuming that your streams topology is stateless. For example going back to the original example,

master topic -> my processing in a mapper/filter -> output topics

"my processing in a mapper/filter" should be stateless. I.e. Not re-partitioning (aka writing to a persistent re-partition topic) or doing a toTable() (aka writing to a changelog topic). If the processing fails further down the topology and you commit the transaction (by following any of the 3 option mention above - flatmap, branch or filter - then you have to cater for manually or programmatically eventually deleting that inconsistent state. That would mean writing extra custom code for automatic this.

I would personally expect Streams to also give you a LogAndSkip option for any unhandled runtime exception, not only for deserialization and production ones.

Has anyone any ideas on this?

Nomography

Answer 6

I don't believe these examples work at all when working with Avro.

When the schema can't be resolved (i.e there is bad/non-avro message corrupting the topic, for example) there is no key or value to deserialize in the first place because by the time the DSL .branch() code is called, the exception has already been thrown (or handled).

Can anyone confirm if this i indeed the case? The very fluent approach you refer to here isn't possible when working with Avro?

KIP-161 does explain how to use a handler, however, it's much more fluent to see it as part of the topology.

Leeds