Method to publish a ProducerRecord

embedded-kafka/src/main/scala/net/manub/embeddedkafka/EmbeddedKafka.scala

@@ -218,6 +218,24 @@ sealed trait EmbeddedKafkaSupport {
                                      serializer),
                    new ProducerRecord[String, T](topic, message))
 
+  /**
+    * Publishes synchronously a message to the running Kafka broker.
+    *
+    * @param topic          the topic to which publish the message (it will be auto-created)
+    * @param producerRecord the producerRecord of type [[T]] to publish
+    * @param config         an implicit [[EmbeddedKafkaConfig]]
+    * @param serializer     an implicit [[Serializer]] for the type [[T]]
+    * @throws KafkaUnavailableException if unable to connect to Kafka
+    */
+  @throws(classOf[KafkaUnavailableException])
+  def publishToKafka[T](topic: String, producerRecord: ProducerRecord[String, T])(
+      implicit config: EmbeddedKafkaConfig,
+      serializer: Serializer[T]): Unit =
+    publishToKafka(new KafkaProducer(baseProducerConfig.asJava,
+                                     new StringSerializer(),
+                                     serializer),
+      producerRecord)
+
   /**
     * Publishes synchronously a message to the running Kafka broker.
     *

embedded-kafka/src/test/scala/net/manub/embeddedkafka/EmbeddedKafkaMethodsSpec.scala

@@ -9,6 +9,7 @@ import org.apache.kafka.clients.producer.{
   ProducerConfig,
   ProducerRecord
 }
+import org.apache.kafka.common.header.internals.RecordHeaders
 import org.apache.kafka.common.serialization.{
   ByteArraySerializer,
   StringDeserializer,
@@ -58,6 +59,33 @@ class EmbeddedKafkaMethodsSpec
       consumer.close()
     }
 
+    "publish synchronously a String message with a header to Kafka" in {
+      implicit val serializer = new StringSerializer()
+      implicit val deserializer = new StringDeserializer()
+      val message = "hello world!"
+      val topic = "publish_test_topic_with_header"
+      val headerValue = "my_header_value"
+      val headers = new RecordHeaders().add("my_header", headerValue.toCharArray.map(_.toByte))
+      val producerRecord = new ProducerRecord[String, String](topic, null, "key", message, headers)
+
+      publishToKafka(topic, producerRecord)
+
+      val consumer = kafkaConsumer
+      consumer.subscribe(List(topic).asJava)
+
+      val records = consumer.poll(consumerPollTimeout)
+
+      records.iterator().hasNext shouldBe true
+      val record = records.iterator().next()
+
+      record.value() shouldBe message
+      val myHeader = record.headers().lastHeader("my_header")
+      val actualHeaderValue = new String(myHeader.value().map(_.toChar))
+      actualHeaderValue shouldBe headerValue
+
+      consumer.close()
+    }
+
     "publish synchronously a String message with String key to Kafka" in {
       implicit val serializer = new StringSerializer()
       implicit val deserializer = new StringDeserializer()