public void onNewBatch(String topic, Cluster cluster, int prevPartition) { // With the introduction of KIP-480 to enhance record production // throughput Kafka's API calls the partition() method twice resulting // in partitions being skipped. More information about this here: // https://issues.apache.org/jira/browse/KAFKA-9965 // The temporary solution is to use the callback method 'onNewBatch' // to decrease the counter to stabilize the round-robin logic. String bucketName = lastBucket.get(); Bucket bucket = buckets.get(bucketName); if (bucket != null) { bucket.decrementCounter(); } lastBucket.remove(); }