Page History

...

We believe that the user should be able to develop custom exception handlers for managing producer exceptions. On the other hand, this will be an expert-level API, and using that may result in strange behaviour in the system, making it hard to find the root cause. Therefore, the custom handler is currently limited to handling RecordTooLargeException and UnknownTopicOrPartitionException. The motivation for this KIP is derived from the following use cases:

• RecordTooLargeException:
  • In transactions, the producer collects multiple records in batches. Then a RecordTooLargeException related to a single record leads to failing the entire batch. A custom exception handler in this case may decide on dropping the record and continuing the processing. See Example 1, please.
  When a user tries to write into
  • More over, in this case, the producer does throw the exception to the user. The users using the producer directly can react to it. However, for Kafka Streams, a record that is too large is a poison pill record, and there is no easy way to skip over it. Currently, Kafka Streams treats this error as fatal and seeks to go back to the last commit of the input topic offset and retry to hit the same error again. It would require a major change inside Kafka Streams to add a bookkeeping code to track this error case correctly and to skip over this record when retrying. A handler would allow us to react to this error inside the producer, i.e., local to where the error happens, and thus simplify the overall code significantly.

• UnknownTopicOrPartitionException: For this case, the producer handles this exception internally and only issues a WARN log about missing metadata and retries internally. Later, when the producer hits the "deliver.timeout.ms", it throws a TimeoutException and the user can only blindly retry, resulting in an infinite retry loop. The thrown TimeoutException "cuts" the connection to the underlying root cause of missing metadata (which could indeed be a transient error but is persistent for a non-existing topic
, it returns a retryable error code; with infinite retries, the producer would hang retrying forever. A custom handler in this case could help
• ). Thus, there is no programmatic way to break the infinite retry loop. Kafka Streams also blindly retries for this case, and the application gets stuck.
  

This KIP introduces an interface that can be implemented by the user to handle the exceptions UnknownTopicOrPartitionException and RecordTooLargeException.
Question: Why do we need an interface for handling the exceptions? Could we have a couple of simple producer configuration options for those two exceptions? 
Answer: We aim at giving the user the flexibility of an interface. For example, facing UnknownTopicOrPartitionException, the user may want to raise an error for some topics but retry it for other topics. Having a configuration option with a fixed set of possibilities does not serve the user's needs. See Example 2, please. 

...

We introduce the ProducerExceptionHandler interface, which can be implemented by the user to manageUnknownTopicOrPartitionException and RecordTooLargeException in the desired manner. It can either stop or continue the processing. Stopping processing is named as FAIL since the transaction or record sending (in non-transactional mode) will fail. In the case of continuing processing, either the record is dropped and the error is ignored (SWALLOW) or sending is retried (RETRY). The accepted responses for RecordTooLargeException are FAIL and SWALLOW. Therefore, RETRY will be interpreted and executed as FAIL.

...