ruanliang-hualun commented on code in PR #12066:
URL: https://github.com/apache/kafka/pull/12066#discussion_r854763466
##########
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:
the drainIndex should be related to tp, take a look at the following
example(node has two tp, tp-0 tp-1)
// drainIndex = 1
drainIndex = getDrainIndex();
do {
drainIndex = 1 ;
tp = tp-1;
// update drainIndex to 0 without updateDrainIndex
drainIndex = 0 ;
synchronized (deque) {
// satisfy the break condition
break;
}
}
while (start != drainIndex)
// record drainIndex = 0; (the drainIndex should be 1 right ? so the
next drainBatch will begin with tp-1)
updateDrainIndex()
--
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
