Versions Compared

Old Version 8

changes.mady.by.user Alieh Saeedi

Saved on

compared with

New Version 9

changes.mady.by.user Alieh Saeedi

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

Proposed Changes

We aim at modifying enabling users to ignore the `send(ProducerRecord)` errors by explicitly calling `flush()` and the `flushcommitTransaction(ignoreError)` method such so that it clears the latest error produced by `send(ProducerRecord)`. Therefore  In order to keep it safe, the change is applied to transactions with no pending record. It mean that, if   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.

...

  • `commitTransaction(ignoreError)` is defined so that if it is set to true, it MAY clear the last error. The default is false.
  • If the user uses `flush()`+ `commitTransaction(true)` then the latest error is ignored and the transaction transits back from the error state. it aaffects only if `flush()` is called beforehand.

Public Interfaces


Code Block
languagejava
titleProducerConfigKafkaProducer
     /**
     * If {@link #flush()} is called explicitly before this method and the input config parameter determines ignoring errors,
     * this method clears the last exception.
.
.
 
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{@link #send(ProducerRecord) + "</code>} call and transits the transaction
     * out of the error state.;

. .
.
static {
CONFIG = new ConfigDef().define(
.....
.
.
.
                         .define(ENABLE_FLUSH_ERROR_CLEARING_CONFIG, Thereupon, the method carries out the same procedures as delineated in {@link #commitTransaction()}.
     *  
     * @param commitOptions The method options
     * @throws IllegalStateException if no transactional.id has been configured or no transaction has been started
     * @throws ProducerFencedException fatal error indicating another producer with the Type.BOOLEAN,
  same transactional.id is active
     * @throws org.apache.kafka.common.errors.UnsupportedVersionException fatal error indicating the broker
     *         does not support transactions  false,
      (i.e. if its version is lower than 0.11.0.0)
     * @throws org.apache.kafka.common.errors.AuthorizationException fatal error indicating that the configured
     *         Importancetransactional.LOW,
id  is  not  authorized.  See  the  exception  for  more  details
          *     ENABLE_FLUSH_ERROR_CLEARING_DOC);
 }
Code Block
languagejava
titleKafkaProducer
....
.
.
.
* 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 org.apache.kafka.common.errors.InvalidProducerEpochException if the producer has attempted to produce with an old epoch
     *         to the partition leader. See the exception for more details
     * @throws KafkaException if the producer has encountered a previous fatal or abortable error, or for any
     *         other unexpected error
     * @throws TimeoutException if the time taken for committing the transaction has surpassed <code>max.block.ms</code>.
     * @throws InterruptException if the thread is interrupted while blocked
     */
     public void commitTransaction(Map<String, ?> commitOptions) throws ProducerFencedException {}

...

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

...

Unit tests for `KafkaProducer` to show that the new feature works with different `send()` errors and exceptions such as RecordTooLargeException.

Rejected Alternatives

  • Just `flush()
  • `send()` throws ApiException
  • Just commitTransaction()`