public List < KafkaTopicPartitionLeader > getPartitionLeadersForTopics ( List < String > topics ) { List < KafkaTopicPartitionLeader > partitions = new LinkedList <> ( ) ; retryLoop : for ( int retry = 0 ; retry < numRetries ; retry ++ ) { brokersLoop : for ( int arrIdx = 0 ; arrIdx < seedBrokerAddresses . length ; arrIdx ++ ) { LOG . info ( "Trying to get topic metadata from broker {} in try {}/{}" , seedBrokerAddresses [ currentContactSeedBrokerIndex ] , retry , numRetries ) ; try { // clear in case we have an incomplete list from previous tries partitions . clear ( ) ; for ( TopicMetadata item : consumer . send ( new TopicMetadataRequest ( topics ) ) . topicsMetadata ( ) ) { if ( item . errorCode ( ) != ErrorMapping . NoError ( ) ) { // warn and try more brokers LOG . warn ( "Error while getting metadata from broker {} to find partitions for {}. Error: {}." , seedBrokerAddresses [ currentContactSeedBrokerIndex ] , topics . toString ( ) , ErrorMapping . exceptionFor ( item . errorCode ( ) ) . getMessage ( ) ) ; useNextAddressAsNewContactSeedBroker ( ) ; continue brokersLoop ; } if ( ! topics . contains ( item . topic ( ) ) ) { LOG . warn ( "Received metadata from topic " + item . topic ( ) + " even though it was not requested. Skipping ..." ) ; useNextAddressAsNewContactSeedBroker ( ) ; continue brokersLoop ; } for ( PartitionMetadata part : item . partitionsMetadata ( ) ) { Node leader = brokerToNode ( part . leader ( ) ) ; KafkaTopicPartition ktp = new KafkaTopicPartition ( item . topic ( ) , part . partitionId ( ) ) ; KafkaTopicPartitionLeader pInfo = new KafkaTopicPartitionLeader ( ktp , leader ) ; partitions . add ( pInfo ) ; } } break retryLoop ; // leave the loop through the brokers } catch ( Exception e ) { //validates seed brokers in case of a ClosedChannelException validateSeedBrokers ( seedBrokerAddresses , e ) ; LOG . warn ( "Error communicating with broker {} to find partitions for {}. {} Message: {}" , seedBrokerAddresses [ currentContactSeedBrokerIndex ] , topics , e . getClass ( ) . getName ( ) , e . getMessage ( ) ) ; LOG . debug ( "Detailed trace" , e ) ; // we sleep a bit. Retrying immediately doesn't make sense in cases where Kafka is reorganizing the leader metadata try { Thread . sleep ( 500 ) ; } catch ( InterruptedException e1 ) { // sleep shorter. } useNextAddressAsNewContactSeedBroker ( ) ; } } // brokers loop } // retries loop return partitions ; }
Send request to Kafka to get partitions for topics .