Current state: "Under Discussion"
Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]
JIRA: here
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Currently, Kafka Streams does not support ProcessingExceptionHandler for GlobalKTable processors. When a processing exception occurs during GlobalKTable record processing, the GlobalStreamThread fails and terminates, causing the entire Kafka Streams application to shut-down. This behavior is inconsistent with regular KStream/KTable processing, where ProcessingExceptionHandler allows applications to handle exceptions gracefully and continue processing.
No changes in public interfaces. We will be reusing existing interfaces.
This KIP extends the applicability of the existing ProcessingExceptionHandler to GlobalKTable processors. Currently, this exception handler only applies to regular stream processing (KStream/KTable). After this KIP, the same handler will also
handle processing exceptions in GlobalKTable.
After this KIP, the ProcessingExceptionHandler configured via `default.processing.exception.handler` will be invoked for:
- KStream processing exceptions (existing)
- KTable processing exceptions (existing)
- GlobalKTable processing exceptions (new capability added by this KIP)
DeserializationExceptionHandler is configured via StreamsConfig, instantiated via TopologyConfig, passed to GlobalStateUpdateTask, and used during GlobalKTable deserialization.
The infrastructure for configuration, instantiation, and passing handlers to global processing already exists. This KIP simply extends it to include ProcessingExceptionHandler alongside the existing DeserializationExceptionHandler. ProcessingExceptionHandler will follow the identical pattern, enabling exception handling during GlobalKTable processor processing.
Add ProcessingExceptionHandler as a constructor parameter to GlobalStateUpdateTask and pass it to processor initialization, mirroring the existing DeserializationExceptionHandler implementation.
In this KIP, DLQ records returned by the handler will be logged but NOT sent to Kafka for GlobalKTable processors. This is because GlobalKTable processing does not currently have producer infrastructure. We will cover DLQ in a separate KIP.
This change is fully backward compatible:
- Applications without a configured handler experience no behavior change
- Applications with a configured handler automatically get GlobalKTable support
- The handler interface is not modified
- No new configuration properties are introduced
Since this is a fairly small change, unit tests and integration test should be sufficient.