Question About How to send kafka message with Header

["Xu, Anna" <an...@sap.com.INVALID>]

Dear,

I want to send kafka message with header. I checked the doc of
https://spark.apache.org/docs/3.0.0-preview/structured-streaming-kafka-integration.html

and found I can add a headers column in the dataDF.  I’m not sure how can I add the headers into the dataDF, I used following method But it failed with errors:

val ds: Dataset[(String, String,Array[AnyRef])] = myData.map(r => (r.getAs[String]("key"),
  r.getAs[String]("value") ,header2.toArray))

[cid:image001.png@01D7D732.2712A930]

Following is my code:

  def main(args:Array[String]):Unit={
    val appName = "testKafka2"
    val mode = "local"
    val sparkSession :SparkSession = getStreamSparkDefault(appName: String,mode: String)
    import sparkSession.implicits._
//    val header1: util.List[Header] = util.Arrays.asList(new RecordHeader("tenant", "mytenant".getBytes())
//      , new RecordHeader("datasource", "c4c".getBytes())
//    )
    val header2: util.List[KafkaMessageHeader] = util.Arrays.asList(new KafkaMessageHeader("tenant", "mytenant".getBytes())
      , new KafkaMessageHeader("datasource", "c4c".getBytes())
    )
    val data = Seq (("iphone", "2007"),("iphone 3G","2008"),
      ("iphone 3GS","2009"),
      ("iphone 4","2010"),
      ("iphone 4S","2011"),
      ("iphone 5","2012"),
      ("iphone 8","2014"),
      ("iphone 10","2017"))
    val myData = sparkSession.createDataFrame(data).toDF("key","value")
    val ds: Dataset[(String, String,Array[AnyRef])] = myData.map(r => (r.getAs[String]("key"),r.getAs[String]("value") ,header2.toArray))
    ds.show(false)

    ds//.selectExpr( "key", "value","partition","topic","headers")
          .write
          .format("kafka")
          .option("kafka.bootstrap.servers", "localhost:9092")
          .option("topic", "testTopic")
          .save()
    println("xxx")
  }

Do you have any idea about that?

Thanks
Anna