code: public boolean setAndCheckDiscoveredPartition(KafkaTopicPartition partition) {
  		if (isUndiscoveredPartition(partition)) {
  			discoveredPartitions.add(partition);
  
  			return KafkaTopicPartitionAssigner.assign(partition, numParallelSubtasks) == indexOfThisSubtask;
  		}
  
  		return false;
  	}
docs: Sets a partition as discovered. Partitions are considered as new
  if its partition id is larger than all partition ids previously
  seen for the topic it belongs to. Therefore, for a set of
  discovered partitions, the order that this method is invoked with
  each partition is important.
  
  <p>If the partition is indeed newly discovered, this method also returns
  whether the new partition should be subscribed by this subtask.
  
  @param partition the partition to set and check
  
  @return {@code true}, if the partition wasn't seen before and should
  be subscribed by this subtask; {@code false} otherwise