code: public KafkaMessage sendMessage(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);
          try {
              RecordMetadata metadata = producer.send(record).get();
              KafkaMessage kafkaMessage = new KafkaMessage(message);
              kafkaMessage.partition(metadata.partition());
              kafkaMessage.offset(metadata.offset());
              return kafkaMessage;
          } catch (InterruptedException | ExecutionException e) {
              throw new KafkaException(e);
          }
      }
docs: Sends a message, specifying {@link ProducerType}.
  
  @param type
  @param message
  @return a copy of message filled with partition number and offset