Сбой десериализации сообщения Spring kafka, поскольку для типа содержимого установлено значение application/json

Я отправляю сообщение Кафке следующим образом:

private KafkaTemplate<String, MyMessage > kafkaTemplate;

public void sendMessage(MyMessage data) {

    Message<MyMessage > message = MessageBuilder
            .withPayload(data)
            .setHeader(KafkaHeaders.TOPIC,topic)
            .setHeader(KafkaHeaders.MESSAGE_KEY,data.getKey())
            .build();
    kafkaTemplate.send(message);

У меня есть собственный де/сериализатор для MyMessage, ключ представляет собой простую строку

Я пытаюсь обработать сообщение следующим образом:

@StreamListener
    public void process(@Input("input") KStream<String,MyMessage> myStream){
            final Serde<String> stringSerde = Serdes.String();
            final MyMessageSerde myMessageSerde = new MyMessageSerde();
            myStream
                    .groupBy((key,value)-> value.getObjectKey(), Serialized.with(stringSerde,myMessageSerde))
                    .aggregate(ArrayList::new,
                            (newKey,val,agg) -> {
                                agg.add(val);
                                return agg;
                            },
                            Materialized.<String, ArrayList<MyMessage>, KeyValueStore<Bytes, byte[]>>as("object-keys")
                                    .withKeySerde(stringSerde)
                                    .withValueSerde(new ArrayListSerde(myMessageSerde)));

    ...
    interface MyStreamProcessor {
        @Input("input")
        KStream<?, ?> input();
    }

Это не удается с ошибкой десериализации, и после отладки я вижу, что причина в том, что заголовок сообщения типа содержимого установлен на application/json, и поэтому он пытается десериализовать сообщение как JSON. Как устанавливается заголовок типа содержимого и как я могу его переопределить? Я не отправляю сообщения в виде JSON, а скорее в виде byte[], как обычные сообщения Kafka, поэтому я хочу, чтобы использовался мой пользовательский де/сериализатор.

Что еще более странно, так это то, что если я поменяю это на KTable, он будет работать нормально.

@StreamListener
public void process(@Input("input") KTable<String,MyMessage> myTable){
        final Serde<String> stringSerde = Serdes.String();
        final MyMessageSerde myMessageSerde = new MyMessageSerde();
        KStream<String,MyMessage> myStream = myTable.toStream();
        myStream
                .groupBy((key,value)-> value.getObjectKey(), Serialized.with(stringSerde,myMessageSerde))
                .aggregate(ArrayList::new,
                        (newKey,val,agg) -> {
                            agg.add(val);
                            return agg;
                        },
                        Materialized.<String, ArrayList<MyMessage>, KeyValueStore<Bytes, byte[]>>as("object-keys")
                                .withKeySerde(stringSerde)
                                .withValueSerde(new ArrayListSerde(myMessageSerde)));

...
interface MyStreamProcessor {
    @Input("input")
    KTable<?, ?> input();
}

Почему с KTable сообщение десериализуется правильно? В чем разница в поведении между KStream и KTable?