test_class.declarator: keyBytes = "key".getBytes()
test_class.modifier: private
test_class.original_string: private byte[] keyBytes = "key".getBytes();
test_class.type: byte[]
test_class.var_name: keyBytes
test_class.declarator: partitioner = new DefaultPartitioner()
test_class.modifier: private
test_class.original_string: private Partitioner partitioner = new DefaultPartitioner();
test_class.type: Partitioner
test_class.var_name: partitioner
test_class.declarator: node0 = new Node(0, "localhost", 99)
test_class.modifier: private
test_class.original_string: private Node node0 = new Node(0, "localhost", 99);
test_class.type: Node
test_class.var_name: node0
test_class.declarator: node1 = new Node(1, "localhost", 100)
test_class.modifier: private
test_class.original_string: private Node node1 = new Node(1, "localhost", 100);
test_class.type: Node
test_class.var_name: node1
test_class.declarator: node2 = new Node(2, "localhost", 101)
test_class.modifier: private
test_class.original_string: private Node node2 = new Node(2, "localhost", 101);
test_class.type: Node
test_class.var_name: node2
test_class.declarator: nodes = new Node[] {node0, node1, node2}
test_class.modifier: private
test_class.original_string: private Node[] nodes = new Node[] {node0, node1, node2};
test_class.type: Node[]
test_class.var_name: nodes
test_class.declarator: topic = "test"
test_class.modifier: private
test_class.original_string: private String topic = "test";
test_class.type: String
test_class.var_name: topic
test_class.declarator: partitions = asList(new PartitionInfo(topic, 1, null, nodes, nodes),
                                                      new PartitionInfo(topic, 2, node1, nodes, nodes),
                                                      new PartitionInfo(topic, 0, node0, nodes, nodes))
test_class.modifier: private
test_class.original_string: private List<PartitionInfo> partitions = asList(new PartitionInfo(topic, 1, null, nodes, nodes),
                                                      new PartitionInfo(topic, 2, node1, nodes, nodes),
                                                      new PartitionInfo(topic, 0, node0, nodes, nodes));
test_class.type: List<PartitionInfo>
test_class.var_name: partitions
test_class.declarator: cluster = new Cluster(asList(node0, node1, node2), partitions, Collections.<String>emptySet())
test_class.modifier: private
test_class.original_string: private Cluster cluster = new Cluster(asList(node0, node1, node2), partitions, Collections.<String>emptySet());
test_class.type: Cluster
test_class.var_name: cluster
test_class.file: clients/src/test/java/org/apache/kafka/clients/producer/internals/DefaultPartitionerTest.java
test_class.identifier: DefaultPartitionerTest
test_case.body: @Test
      public void testRoundRobinWithUnavailablePartitions() {
          // When there are some unavailable partitions, we want to make sure that (1) we always pick an available partition,
          // and (2) the available partitions are selected in a round robin way.
          int countForPart0 = 0;
          int countForPart2 = 0;
          for (int i = 1; i <= 100; i++) {
              int part = partitioner.partition("test", null, null, null, null, cluster);
              assertTrue("We should never choose a leader-less node in round robin", part == 0 || part == 2);
              if (part == 0)
                  countForPart0++;
              else
                  countForPart2++;
          }
          assertEquals("The distribution between two available partitions should be even", countForPart0, countForPart2);
      }
test_case.class_method_signature: DefaultPartitionerTest.testRoundRobinWithUnavailablePartitions()
test_case.full_signature: @Test public void testRoundRobinWithUnavailablePartitions()
test_case.identifier: testRoundRobinWithUnavailablePartitions
test_case.invocations: partition assertTrue assertEquals
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testRoundRobinWithUnavailablePartitions()
focal_class.declarator: counter = new AtomicInteger(new Random().nextInt())
focal_class.modifier: private final
focal_class.original_string: private final AtomicInteger counter = new AtomicInteger(new Random().nextInt());
focal_class.type: AtomicInteger
focal_class.var_name: counter
focal_class.file: clients/src/main/java/org/apache/kafka/clients/producer/internals/DefaultPartitioner.java
focal_class.identifier: DefaultPartitioner
focal_class.interfaces: implements Partitioner
focal_class.class_method_signature: DefaultPartitioner.toPositive(int number)
focal_class.full_signature: private static int toPositive(int number)
focal_class.identifier: toPositive
focal_class.modifiers: private static
focal_class.parameters: (int number)
focal_class.return: int
focal_class.signature: int toPositive(int number)
focal_class.class_method_signature: DefaultPartitioner.main(String[] args)
focal_class.full_signature: public static void main(String[] args)
focal_class.identifier: main
focal_class.modifiers: public static
focal_class.parameters: (String[] args)
focal_class.return: void
focal_class.signature: void main(String[] args)
focal_class.class_method_signature: DefaultPartitioner.configure(Map<String, ?> configs)
focal_class.full_signature: public void configure(Map<String, ?> configs)
focal_class.identifier: configure
focal_class.modifiers: public
focal_class.parameters: (Map<String, ?> configs)
focal_class.return: void
focal_class.signature: void configure(Map<String, ?> configs)
focal_class.class_method_signature: DefaultPartitioner.partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster)
focal_class.full_signature: public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster)
focal_class.identifier: partition
focal_class.modifiers: public
focal_class.parameters: (String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster)
focal_class.return: int
focal_class.signature: int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster)
focal_class.class_method_signature: DefaultPartitioner.close()
focal_class.full_signature: public void close()
focal_class.identifier: close
focal_class.modifiers: public
focal_class.parameters: ()
focal_class.return: void
focal_class.signature: void close()
focal_method.body: public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
          List<PartitionInfo> partitions = cluster.partitionsForTopic(topic);// 先从集群中取出所有分区
          int numPartitions = partitions.size();
          if (keyBytes == null) {            // 没有key的情况走这个分支
              int nextValue = counter.getAndIncrement(); // 计数服务+1
              List<PartitionInfo> availablePartitions = cluster.availablePartitionsForTopic(topic);// 取出可用的分区
              if (availablePartitions.size() > 0) {
                  int part = DefaultPartitioner.toPositive(nextValue) % availablePartitions.size();
                  return availablePartitions.get(part)
                                            .partition();
              } else {
                  // no partitions are available, give a non-available partition
                  // 没有可用的分区，只能返回一个不可用的分区
                  return DefaultPartitioner.toPositive(nextValue) % numPartitions;
              }
          } else {
              // hash the keyBytes to choose a partition
              return DefaultPartitioner.toPositive(Utils.murmur2(keyBytes)) % numPartitions;
          }
      }
focal_method.class_method_signature: DefaultPartitioner.partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster)
focal_method.full_signature: public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster)
focal_method.identifier: partition
focal_method.invocations: partitionsForTopic size getAndIncrement availablePartitionsForTopic size toPositive size partition get toPositive toPositive murmur2
focal_method.modifiers: public
focal_method.parameters: (String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster)
focal_method.return: int
focal_method.signature: int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster)