Status

Current state: "Under Discussion"

Discussion thread: here 

JIRA: KAFKA-15259 - Getting issue details... STATUS

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

In transactions, having a poison record or sending the record to a topic with issues ends up going to an error state, which leads to dropping the whole batch. This logic does not allow the user to do advanced error handling and ignore some errors. More specifically, the custom production exception handler of Kafka Streams can not influence the control to skip the error by dropping the bad record and committing the transaction with the rest of the batch records.  

Proposed Changes

We aim at modifying the `flush()` method such that it clears the latest error produced by `send(ProducerRecord)`. Therefore, if the user calls `flush()` after a `send`, the error produced by send (if existing) is cleared. 

It should be noted that if `flush()` is called explicitly before `commitTransaction()`, this method will NOT throw any exception related to the latest `send(ProducerRecord)` call., since `flush()` clears the last received exception and transits the transaction out of the error state.

In order to preserve the former functionalities, we define a new producer config parameter called enable.flush.error.clearing. The default value is false and if it is set to true, the error clearing of flush is applied. 

Public Interfaces

.
.
.
 
public static final String ENABLE_FLUSH_ERROR_CLEARING_CONFIG = "enable.flush.error.clearing";
private static final String ENABLE_FLUSH_ERROR_CLEARING_DOC = "When set to true, <code>" + flush() + "</code> clears the error produced by the latest <code>" + send(ProducerRecord) + "</code> and transits the transaction out of the error state.;

. .
.
static {
CONFIG = new ConfigDef().define(
.....
.
.
.
                         .define(ENABLE_FLUSH_ERROR_CLEARING_CONFIG,
                                 Type.BOOLEAN,
                                 false,
                                 Importance.LOW,
                                 ENABLE_FLUSH_ERROR_CLEARING_DOC);
 }

....
.
.
.
* Additionally, this method clears the last exception in the transaction and transits the transaction out of error state.
.
.
@Override
public void flush() {}.
.
.
.
* It should be noted that if <code>flush()</code> is called explicitly beforehand, this method will NOT throw any
* exception related to the {@link #send(ProducerRecord)} calls. Since <code>flush()</code> clears the last received
* exception and transits the transaction out of error state.
.
.
.
public void commitTransaction() throws ProducerFencedException {}

Compatibility, Deprecation, and Migration Plan

Since the default behaviour is preserved by default false value of the newly defined config parameter, the change has no impact on existing users.

Test Plan

Unit tests for `KafkaProducer` to show that the new feature works with send errors such RecordTooLargeException and other errors.