code: public static List<KafkaPartition> getPartitions(State state) {
      List<KafkaPartition> partitions = Lists.newArrayList();
      String topicName = state.getProp(KafkaSource.TOPIC_NAME);
      for (int i = 0;; i++) {
        if (!state.contains(KafkaUtils.getPartitionPropName(KafkaSource.PARTITION_ID, i))) {
          break;
        }
        KafkaPartition partition = new KafkaPartition.Builder().withTopicName(topicName)
            .withId(state.getPropAsInt(KafkaUtils.getPartitionPropName(KafkaSource.PARTITION_ID, i)))
            .withLeaderId(state.getPropAsInt(KafkaUtils.getPartitionPropName(KafkaSource.LEADER_ID, i)))
            .withLeaderHostAndPort(state.getProp(KafkaUtils.getPartitionPropName(KafkaSource.LEADER_HOSTANDPORT, i)))
            .build();
        partitions.add(partition);
      }
      if (partitions.isEmpty()) {
        partitions.add(getPartition(state));
      }
      return partitions;
    }
docs: Get a list of {@link KafkaPartition}s from a {@link State} object. The given {@link State} should contain property
  {@link KafkaSource#TOPIC_NAME}. If there are multiple partitions in the {@link State}, all partitions should have
  the same topic name.
  
  It first checks whether the given {@link State} contains "partition.id.i", "leader.id.i" and
  "leader.hostandport.i", i = 0,1,2,...
  
  Otherwise it will call {@link #getPartition(State)}.