artemlivshits commented on code in PR #12462:
URL: https://github.com/apache/kafka/pull/12462#discussion_r1063886879
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java:
##########
@@ -287,12 +291,16 @@ public RecordAppendResult append(String topic,
}
}
- // we don't have an in-progress record batch try to allocate a
new batch
- if (abortOnNewBatch) {
- // Return a result that will cause another call to append.
+ // noDqForPartition is true either when 1. partition was
encountered for first time so no Deque existed previously.
+ // 2. DQ was removed due to - all batches were cleared due to
expiration or sender cleared batches after draining.
+ // if so, abort and look to call partitioner -> onNewBatch and
select other partition.
+ // This prevents a single partition getting re-selected after
recent drain.
+ if (abortOnNewBatch && noDqForPartition) {
Review Comment:
`onNewBatch` is now deprecated, so hopefully, people would take a hint not
to override it :-)
But, yeah, agree with calling it creative (I wouldn't have come up with that
given my C++ background -- these ideas are blocked before they appear in my
mind).
--
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.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
