[GitHub] [kafka] cmccabe commented on a change in pull request #8690: KAFKA-9965: Fix uneven distribution in RoundRobinPartitioner

[GitBox <gi...@apache.org>]

cmccabe commented on a change in pull request #8690:
URL: https://github.com/apache/kafka/pull/8690#discussion_r428998676



##########
File path: clients/src/main/java/org/apache/kafka/clients/producer/RoundRobinPartitioner.java
##########
@@ -65,12 +65,20 @@ public int partition(String topic, Object key, byte[] keyBytes, Object value, by
     }
 
     private int nextValue(String topic) {
-        AtomicInteger counter = topicCounterMap.computeIfAbsent(topic, k -> {
-            return new AtomicInteger(0);
-        });
+        AtomicInteger counter = topicCounterMap.
+            computeIfAbsent(topic, k -> new AtomicInteger(0));
         return counter.getAndIncrement();
     }
 
+    @Override
+    public void onNewBatch(String topic, Cluster cluster, int prevPartition) {
+        // After onNewBatch is called, we will call partition() again.
+        // So 'rewind' the counter for this topic.
+        AtomicInteger counter = topicCounterMap.
+            computeIfAbsent(topic, k -> new AtomicInteger(0));
+        counter.getAndDecrement();

Review comment:
       It's possible.  There's no easy fix, though, other than rethinking the `Partitioner` API.  That would be an incompatible change.




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org