showuon commented on code in PR #12066:
URL: https://github.com/apache/kafka/pull/12066#discussion_r854746909
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java:
##########
@@ -559,13 +559,14 @@ private List<ProducerBatch>
drainBatchesForOneNode(Cluster cluster, Node node, i
int size = 0;
List<PartitionInfo> parts = cluster.partitionsForNode(node.id());
List<ProducerBatch> ready = new ArrayList<>();
- /* to make starvation less likely this loop doesn't start at 0 */
+ /* to make starvation less likely each node has it's own drainIndex */
+ int drainIndex = getDrainIndex(node.idString());
int start = drainIndex = drainIndex % parts.size();
do {
PartitionInfo part = parts.get(drainIndex);
TopicPartition tp = new TopicPartition(part.topic(),
part.partition());
- this.drainIndex = (this.drainIndex + 1) % parts.size();
-
+ updateDrainIndex(node.idString(), drainIndex);
Review Comment:
I still think we can updateDrainIndex after while loop. After all, we only
use the `drainIndex` variable inside while loop, not related to
`nodesDrainIndex` map, right?
--
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
