test_class.declarator: assignor = new RangeAssignor()
test_class.modifier: private
test_class.original_string: private RangeAssignor assignor = new RangeAssignor();
test_class.type: RangeAssignor
test_class.var_name: assignor
test_class.file: clients/src/test/java/org/apache/kafka/clients/consumer/RangeAssignorTest.java
test_class.identifier: RangeAssignorTest
test_case.body: @Test
      public void testOneConsumerNoTopic() {
          String consumerId = "consumer";
  
          Map<String, Integer> partitionsPerTopic = new HashMap<>();
  
          Map<String, List<TopicPartition>> assignment = assignor.assign(partitionsPerTopic,
                  Collections.singletonMap(consumerId, Collections.<String>emptyList()));
  
          assertEquals(Collections.singleton(consumerId), assignment.keySet());
          assertTrue(assignment.get(consumerId).isEmpty());
      }
test_case.class_method_signature: RangeAssignorTest.testOneConsumerNoTopic()
test_case.full_signature: @Test public void testOneConsumerNoTopic()
test_case.identifier: testOneConsumerNoTopic
test_case.invocations: assign singletonMap emptyList assertEquals singleton keySet assertTrue isEmpty get
test_case.modifiers: @Test public
test_case.parameters: ()
test_case.return: void
test_case.signature: void testOneConsumerNoTopic()
focal_class.file: clients/src/main/java/org/apache/kafka/clients/consumer/RangeAssignor.java
focal_class.identifier: RangeAssignor
focal_class.class_method_signature: RangeAssignor.name()
focal_class.full_signature: @Override public String name()
focal_class.identifier: name
focal_class.modifiers: @Override public
focal_class.parameters: ()
focal_class.return: String
focal_class.signature: String name()
focal_class.class_method_signature: RangeAssignor.consumersPerTopic(Map<String/* memberId */, List<String/* topic */>> consumerMetadata)
focal_class.full_signature: private Map<String/* topic */, List<String/* memberId */>> consumersPerTopic(Map<String/* memberId */, List<String/* topic */>> consumerMetadata)
focal_class.identifier: consumersPerTopic
focal_class.modifiers: private
focal_class.parameters: (Map<String/* memberId */, List<String/* topic */>> consumerMetadata)
focal_class.return: Map<String/* topic */, List<String/* memberId */>>
focal_class.signature: Map<String/* topic */, List<String/* memberId */>> consumersPerTopic(Map<String/* memberId */, List<String/* topic */>> consumerMetadata)
focal_class.class_method_signature: RangeAssignor.assign(Map<String/* topic */, Integer> partitionsPerTopic,
          Map<String/* memberId */, List<String/* topic */>> subscriptions)
focal_class.full_signature: @Override public Map<String/* memberId */, List<TopicPartition>> assign(Map<String/* topic */, Integer> partitionsPerTopic,
          Map<String/* memberId */, List<String/* topic */>> subscriptions)
focal_class.identifier: assign
focal_class.modifiers: @Override public
focal_class.parameters: (Map<String/* topic */, Integer> partitionsPerTopic,
          Map<String/* memberId */, List<String/* topic */>> subscriptions)
focal_class.return: Map<String/* memberId */, List<TopicPartition>>
focal_class.signature: Map<String/* memberId */, List<TopicPartition>> assign(Map<String/* topic */, Integer> partitionsPerTopic,
          Map<String/* memberId */, List<String/* topic */>> subscriptions)
focal_class.superclass: extends AbstractPartitionAssignor
focal_method.body: @Override
      public Map<String/* memberId */, List<TopicPartition>> assign(Map<String/* topic */, Integer> partitionsPerTopic,
          Map<String/* memberId */, List<String/* topic */>> subscriptions) {
  
          // 将原本的 member订阅了哪几个topic，转换为topic 被哪几个member订阅了 topicPerConsumer => consumerPerTopic
          Map<String/* topic */, List<String/* memberId */>> consumersPerTopic = consumersPerTopic(subscriptions);
  
          // 为每个 member new一个空的TopicPartition订阅集合
          Map<String/* memberId */, List<TopicPartition>> assignment = new HashMap<>();
          for (String memberId : subscriptions.keySet())
              assignment.put(memberId, new ArrayList<TopicPartition>());
  
          // 开始分配
          for (Map.Entry<String, List<String>> topicEntry : consumersPerTopic.entrySet()) {
              String topic = topicEntry.getKey();
              List<String/* memberId */> consumersForTopic = topicEntry.getValue();
  
              // 根据topic名字从 partitionsPerTopic 到partition下有多少个topic
              Integer numPartitionsForTopic = partitionsPerTopic.get(topic);
              if (numPartitionsForTopic == null) {
                  continue;
              }
  
              Collections.sort(consumersForTopic);
  
              int numConsumersForConsumer = consumersForTopic.size();
  
              int N_numPartitionsPerConsumer = numPartitionsForTopic / numConsumersForConsumer;
              int M_consumersWithExtraPartition = numPartitionsForTopic % numConsumersForConsumer;
  
              List<TopicPartition> partitions = AbstractPartitionAssignor.partitions(topic, numPartitionsForTopic);
  
              // n = 分区数/消费者，m = 分区数%消费者，前m个消费者分配n+1个分区（每有一个余数，分配多一个），
              // 后面的消费者消费n个分区
              for (int i = 0, n = consumersForTopic.size(); i < n; i++) {
                  int start = N_numPartitionsPerConsumer * i + Math.min(i, M_consumersWithExtraPartition);
                  int length = N_numPartitionsPerConsumer + (i + 1 > M_consumersWithExtraPartition ? 0 : 1);
                  assignment.get(consumersForTopic.get(i))
                            .addAll(partitions.subList(start, start + length));
              }
          }
          return assignment;
      }
focal_method.class_method_signature: RangeAssignor.assign(Map<String/* topic */, Integer> partitionsPerTopic,
          Map<String/* memberId */, List<String/* topic */>> subscriptions)
focal_method.full_signature: @Override public Map<String/* memberId */, List<TopicPartition>> assign(Map<String/* topic */, Integer> partitionsPerTopic,
          Map<String/* memberId */, List<String/* topic */>> subscriptions)
focal_method.identifier: assign
focal_method.invocations: consumersPerTopic keySet put entrySet getKey getValue get sort size partitions size min addAll get get subList
focal_method.modifiers: @Override public
focal_method.parameters: (Map<String/* topic */, Integer> partitionsPerTopic,
          Map<String/* memberId */, List<String/* topic */>> subscriptions)
focal_method.return: Map<String/* memberId */, List<TopicPartition>>
focal_method.signature: Map<String/* memberId */, List<TopicPartition>> assign(Map<String/* topic */, Integer> partitionsPerTopic,
          Map<String/* memberId */, List<String/* topic */>> subscriptions)