code: 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);
              }
          }
      }
docs: 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
  
  @param ppd the producer pool request object