...

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 the 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()` code method such that it clears the latest error produced by `send(ProducerRecord)`. Therefore,  if if the user calls `flush()` after a `send`, the error produced by send (if existing) is cleared. and ...... 

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 , 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 worksis 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);
 }

...