code: public Future<KafkaMessage> sendMessageAsync(ProducerType type, KafkaMessage message) {
          String key = message.key();
          ProducerRecord<String, byte[]> record = StringUtils.isEmpty(key)
                  ? new ProducerRecord<>(message.topic(), message.content())
                  : new ProducerRecord<>(message.topic(), key, message.content());
          KafkaProducer<String, byte[]> producer = getJavaProducer(type);
          Future<RecordMetadata> fRecordMetadata = producer.send(record);
          FutureTask<KafkaMessage> result = new FutureTask<>(new Callable<KafkaMessage>() {
              @Override
              public KafkaMessage call() throws Exception {
                  RecordMetadata recordMetadata = fRecordMetadata.get();
                  if (recordMetadata != null) {
                      KafkaMessage kafkaMessage = new KafkaMessage(message);
                      kafkaMessage.partition(recordMetadata.partition());
                      kafkaMessage.offset(recordMetadata.offset());
                      return kafkaMessage;
                  }
                  throw new KafkaException("Error while sending message to broker");
              }
          });
          submitTask(result);
          return result;
      }
docs: Sends a message asynchronously, specifying {@link ProducerType}.
  
  @param type
  @param message
  @return a copy of message filled with partition number and offset