showuon commented on code in PR #12066:
URL: https://github.com/apache/kafka/pull/12066#discussion_r854734845
##########
clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java:
##########
@@ -559,13 +560,12 @@ 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();
-
// Only proceed if the partition has no in-flight batches.
if (isMuted(tp))
continue;
Review Comment:
Have a 2nd look, I found this change is not right. If the tp is muted or
other reason, it `continued`, and we'll have a infinite loop because the
drainIdex won't get incremented, right? Please fix it and add a test for this
case. Thanks.
--
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
