content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License. You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.kafka.clients.producer;
  
  import org.apache.kafka.common.Cluster;
  import org.apache.kafka.common.PartitionInfo;
  import org.apache.kafka.common.utils.Utils;
  
  import java.util.List;
  import java.util.Map;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.ConcurrentMap;
  import java.util.concurrent.atomic.AtomicInteger;
  
  /**
   * The "Round-Robin" partitioner
   * 
   * This partitioning strategy can be used when user wants 
   * to distribute the writes to all partitions equally. This
   * is the behaviour regardless of record key hash. 
   *
   */
  public class RoundRobinPartitioner implements Partitioner {
      private final ConcurrentMap<String, AtomicInteger> topicCounterMap = new ConcurrentHashMap<>();
  
      public void configure(Map<String, ?> configs) {}
  
      /**
       * Compute the partition for the given record.
       *
       * @param topic The topic name
       * @param key The key to partition on (or null if no key)
       * @param keyBytes serialized key to partition on (or null if no key)
       * @param value The value to partition on or null
       * @param valueBytes serialized value to partition on or null
       * @param cluster The current cluster metadata
       */
      @Override
      public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
          int nextValue = nextValue(topic);
          List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);
          if (!availablePartitions.isEmpty()) {
              int part = Utils.toPositive(nextValue) % availablePartitions.size();
              return availablePartitions.get(part).partition();
          } else {
              // no partitions are available, give a non-available partition
              int numPartitions = cluster.partitionsForTopic(topic).size();
              return Utils.toPositive(nextValue) % numPartitions;
          }
      }
  
      private int nextValue(String topic) {
          AtomicInteger counter = topicCounterMap.computeIfAbsent(topic, k -> new AtomicInteger(0));
          return counter.getAndIncrement();
      }
  
      public void close() {}
  }
  