ijuma commented on code in PR #12462:
URL: https://github.com/apache/kafka/pull/12462#discussion_r1063884499
##########
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:
Using reflection is a creative option, but it does mean that if someone
overrides the method to eg add a log, then suddenly things will break. Which
would be quite surprising (it goes without saying).
--
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
