DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
We aim at enabling users to ignore the `send(ProducerRecord)` errors by explicitly calling `flush()` and the `commitTransaction(ignoreError)` method so that it defining a new `commitTransaction` method. The new method clears the latest error 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 pending unflushed record. It mean that, if Therefore, if Therefore, the user calls must explicitly call `flush()` after a `send`, the error produced by send (if existing) is cleared.
- `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.
...
before calling the newly defined `commitTransaction`.
Public Interfaces
If the user passes the `ignore.errors` with the value `true` (in `commitOptions`) to the method, it commits the transaction even in the presence of `send()` errors.
| Code Block | ||||
|---|---|---|---|---|
| ||||
/**
* If {@link #flush()} is called explicitly before this method and the input config parameter determines ignoring errors,
* this method clears the last exception produced by the {@link #send(ProducerRecord)} call and transits the transaction
* out of error state. 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 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(Map<String, ?> commitOptions) throws ProducerFencedException {} |
...
Unit tests for `KafkaProducer` to show that the new feature works with different `send()` errors and exceptions such as RecordTooLargeException.
Rejected Alternatives
- Just Add the feature of clearing errors to `flush()`: `flush()` is not necessarily a transactional method.
- `send()` throws ApiException: may break backward compatibility.
- No need of explicit `flush()` before calling `Just
commitTransaction(commitOptions)`: not safe.