Versions Compared

Old Version 15

changes.mady.by.user Alieh Saeedi

Saved on

compared with

New Version 16

changes.mady.by.user Alieh Saeedi

Saved on

Key

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

...

We aim at enabling users to ignore the `send(ProducerRecord)` errors by defining a new `commitTransaction` method. The new method clears the errors produced by `send(ProducerRecord)` and transits the transaction back from the error state.  In order to keep it safe, the change is applied to transactions with no unflushed record. Therefore, the user must explicitly call `flush()` before calling the newly defined  `commitTransaction`.adding an input parameter to the `send` method so that the user is able to determine not going to the error state by a poison pill record. 

Public Interfaces

If the user 1) is performing a transaction and 2) passes the `CommitOption` `TxnSendOption` with the value `CLEAR`IGNORE_SEND_ERRORS` to the `send` method, it commits the transaction even in the presence of `send()` errors, any poison pill record is excluded from the batch, and the transaction is committed successfully.


Code Block
languagejava
titleKafkaProducer
     /**
     *public This method should only be called if there are no pending writes, i.e., only after calling {@link #flush()}.
     * If there are any errors in sending messages to topics, these errors can be cleared by passing {@link CommitOption#CLEAR_SEND_ERRORS},
     * allowing the transaction to be committed even in case of data loss.
     * <p>
     * If this method is used while there are pending sends, the send errors cannot be cleared.
      
     * @param commitOption The method option
     *
     * @throws IllegalStateException if no transactional.id has been configured or no transaction has been started
     * @throws ProducerFencedException fatal error indicating another producer with the same transactional.id is active
     * @throws org.apache.kafka.common.errors.UnsupportedVersionException fatal error indicating the broker
     *         does not support transactions (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
     *         transactional.id is not authorized. See the exception for more details
     * @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(CommitOption option) throws ProducerFencedException {}

Future<RecordMetadata> send(ProducerRecord<K, V> record, Callback callback, TxnSendOption option) {}


     public enum CommitOptionTxnSendOption {
        /**
         * CommitsThe theirrecoverable ongoing transaction, flushing any unsent records before actually committing
         * the transaction. If any of the records sent in this transaction hit unrecoverable errors,
         * the transaction will not be committed.{@link #send(ProducerRecord)} errors lead the transaction to the error state, which ends up an unsuccessful commit. 
         */
        NONE,
        /**
         * CommitsThe therecords ongoing transaction, first clearing any errors from records already sent in
         * this transaction. If there are any unsent records flushed by this operation which hit 
         * unrecoverable errors, these errors will not be cleared causing irrecoverable errors are excluded from the batch and the transaction willis not be 
         * committed.
         * <p>committed successfully. 	
         * ToNote ensureto thereuse arethis, noonly unsentin recordstransactions, you mustotherwise call {@link #flush(#send(ProducerRecord, Callback, TxnSendOption)} before
         * committing the transaction.
         throws exception
         */
        CLEAR_SEND_ERRORS
    }


We define the method in the interface as well. The `default` implementation helps with backward compatibility.

Code Block
languagejava
titleProducer
     /**
     * See {@link KafkaProducer#commitTransaction(KafkaProducer.CommitOptionKafkaProducer#send(ProducerRecord, Callback, SendTxnOption)}
        */
    default void commitTransaction(KafkaProducer.CommitOptionsend(ProducerRecord record, Callback callback, SendTxnOption option) throws ProducerFencedException {}

...

  • Add the feature of clearing errors to `flush()`: `flush()` is not necessarily a transactional method + `AddPartition` is not done successfully in the next `send`.
  • Add the feature of clearing errors to `commitTxn()`: `flush()` is not necessarily a transactional method + `AddPartition` is not done successfully in the next `send`.
  • `send()` throws ApiException: may break backward compatibility.
  • No need of explicit `flush()` before calling `commitTransaction(commitOptions)` :  not safe .+ `AddPartition` is not done successfully in the next `send`