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 ) ; } }
Adds a list of newly discovered partitions to the fetcher for consuming .