docs: public void addDiscoveredPartitions(List<KafkaTopicPartition> newPartitions) throws IOException, ClassNotFoundException {
  		List<KafkaTopicPartitionState<KPH>> newPartitionStates = createPartitionStateHolders(
  				newPartitions,
  				KafkaTopicPartitionStateSentinel.EARLIEST_OFFSET,
  				timestampWatermarkMode,
  				watermarksPeriodic,
  				watermarksPunctuated,
  				userCodeClassLoader);
  
  		if (useMetrics) {
  			registerOffsetMetrics(consumerMetricGroup, newPartitionStates);
  		}
  
  		for (KafkaTopicPartitionState<KPH> newPartitionState : newPartitionStates) {
  			// the ordering is crucial here; first register the state holder, then
  			// push it to the partitions queue to be read
  			subscribedPartitionStates.add(newPartitionState);
  			unassignedPartitionsQueue.add(newPartitionState);
  		}
  	}