guozhangwang commented on a change in pull request #10445:
URL: https://github.com/apache/kafka/pull/10445#discussion_r605980981
##########
File path:
clients/src/main/java/org/apache/kafka/clients/producer/internals/ProducerBatch.java
##########
@@ -181,20 +212,36 @@ public boolean isDone() {
*
* @param baseOffset The base offset of the messages assigned by the server
* @param logAppendTime The log append time or -1 if CreateTime is being
used
- * @param exception The exception that occurred (or null if the request
was successful)
+ * @param topLevelException The exception that occurred (or null if the
request was successful)
+ * @param recordExceptionMap Record exception map keyed by batchIndex
which overrides `topLevelException`
+ * for records present in the map
* @return true if the batch was completed successfully and false if the
batch was previously aborted
*/
- public boolean done(long baseOffset, long logAppendTime, RuntimeException
exception) {
- final FinalState tryFinalState = (exception == null) ?
FinalState.SUCCEEDED : FinalState.FAILED;
-
+ private boolean done(
+ long baseOffset,
+ long logAppendTime,
+ RuntimeException topLevelException,
+ Map<Integer, RuntimeException> recordExceptionMap
+ ) {
+ final FinalState tryFinalState = (topLevelException == null) ?
FinalState.SUCCEEDED : FinalState.FAILED;
if (tryFinalState == FinalState.SUCCEEDED) {
log.trace("Successfully produced messages to {} with base offset
{}.", topicPartition, baseOffset);
} else {
- log.trace("Failed to produce messages to {} with base offset {}.",
topicPartition, baseOffset, exception);
+ log.trace("Failed to produce messages to {} with base offset {}.",
topicPartition, baseOffset, topLevelException);
}
if (this.finalState.compareAndSet(null, tryFinalState)) {
- completeFutureAndFireCallbacks(baseOffset, logAppendTime,
exception);
+ Function<Integer, RuntimeException> recordExceptions = null;
+ if (topLevelException != null) {
+ recordExceptions = batchIndex -> {
+ if (recordExceptionMap == null) {
+ return topLevelException;
+ } else {
+ return recordExceptionMap.getOrDefault(batchIndex,
topLevelException);
Review comment:
Sounds good, I like this one.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
