public void send ( ProducerPoolData < V > ppd ) { if ( logger . isDebugEnabled ( ) ) { logger . debug ( "send message: " + ppd ) ; } if ( sync ) { Message [ ] messages = new Message [ ppd . data . size ( ) ] ; int index = 0 ; for ( V v : ppd . data ) { messages [ index ] = serializer . toMessage ( v ) ; index ++ ; } ByteBufferMessageSet bbms = new ByteBufferMessageSet ( config . getCompressionCodec ( ) , messages ) ; ProducerRequest request = new ProducerRequest ( ppd . topic , ppd . partition . partId , bbms ) ; SyncProducer producer = syncProducers . get ( ppd . partition . brokerId ) ; if ( producer == null ) { throw new UnavailableProducerException ( "Producer pool has not been initialized correctly. " + "Sync Producer for broker " + ppd . partition . brokerId + " does not exist in the pool" ) ; } producer . send ( request . topic , request . partition , request . messages ) ; } else { AsyncProducer < V > asyncProducer = asyncProducers . get ( ppd . partition . brokerId ) ; for ( V v : ppd . data ) { asyncProducer . send ( ppd . topic , v , ppd . partition . partId ) ; } } }
selects either a synchronous or an asynchronous producer for the specified broker id and calls the send API on the selected producer to publish the data to the specified broker partition