Spark Structured Streaming Kafka Headers property is null in ForEachWriter object

I currently have a ForEachWriter which works on a typed Dataset coming from Kafka. All properties except the headers for each message appear properly inside the ForEachWriter. Code to replicate is at the bottom.

The Schema is exactly as it comes from the DataStreamReader for format Kafka.

root
 |-- topic: string (nullable = true)
 |-- key: string (nullable = true)
 |-- value: string (nullable = true)
 |-- timestamp: string (nullable = true)
 |-- headers: array (nullable = true)
 |    |-- element: struct (containsNull = true)
 |    |    |-- key: string (nullable = true)
 |    |    |-- value: binary (nullable = true)

The Headers are clearly in the Dataset column being printed.

+------------+--------+-------+-----------------------+----------------------------------+
|topic       |key     |value  |timestamp              |headers                           |
+------------+--------+-------+-----------------------+----------------------------------+
|controlTopic|        |TEXT   |2020-11-24 16:07:33.09 |[[SAMPLE, TEXT]]                  |
|controlTopic|        |2Header|2020-11-24 16:43:46.755|[[HEADER1, VAL1], [HEADER2, VAL2]]|
+------------+--------+-------+-----------------------+----------------------------------+

And the appropriate number of KeyValue objects even appear in the Header property of the object, but their values are null. For example the first message will display one element in the List (image to the left) and the second message will display two elements in the List (image to the right).

The code is as follows:

KafkaMessage.java

public class KafkaMessage {

    public String topic;

    public String key;

    public String value;

    public String timestamp;

    public List<KV> headers;

    public KafkaMessage() {
    }

    public KafkaMessage(String topic, String key, String value, String timestamp, List<KV> headers) {

        this.topic = topic;
        this.key = key;
        this.value = value;
        this.timestamp = timestamp;
        this.headers = headers;
    }
}

KV.java

public class KV {
    public String key;
    public byte[] value;

    public KV() {
    }

    public KV(String key, byte[] value) {
        this.key = key;
        this.value = value;
    }
}

MySparkWriter.java

public class MySparkWriter extends ForeachWriter<KafkaMessage> {

    @Override
    public boolean open(long partitionId, long epochId) {
        // open connection, not needed in this example
        return true;
    }

    @Override
    public void process(KafkaMessage kafkaMessage) {
        // here headers will not be null, if there are headers
        List<KV> headers = kafkaMessage.headers;
        for (KV kv : headers) {
            System.out.println(kv.key != null); // kv.key will always be null, will only print in local mode
        }
    }

    @Override
    public void close(Throwable errorOrNull) {
        // close connection, not needed in this example
    }
}

MySparkApp.java Note this is the only class with some code omitted, all you need is to create a Spark Context and call startSpark()

public class MySparkApp {

    ...
    
    public StreamingQuery startSpark() throws TimeoutException {
        // Get Input
        DataStreamReader dsReader = this.sparkSession.readStream()
                .format("kafka")
                .option("kafka.bootstrap.servers", "localhost:9092")
                .option("includeHeaders", true)
                .option("subscribe", "controlTopic");

        Dataset<KafkaMessage> messages = dsReader.load()
                .selectExpr("CAST(topic AS STRING)", "CAST(key AS STRING)", "CAST(value AS STRING)", "CAST(timestamp AS STRING)", "headers")
                .as(ExpressionEncoder.javaBean(KafkaMessage.class));

        // debug prints
        messages.writeStream()
            .format("console")
            .outputMode("update")
            .trigger(Trigger.ProcessingTime("2 seconds"))
            .option("truncate", "false")
            .start();
        messages.printSchema();

        // Start Streaming
        return messages
                .writeStream()
                .trigger(Trigger.ProcessingTime("2 seconds"))
                .foreach(this.sparkWriter) // this is a MySparkWriter instance
                .option("checkpointLocation", "/checkpoint/"+ sparkSession.sparkContext().applicationId())
                .start();
    }
    
    ...

}