Kafka Spring Integration：标题不是为Kafka消费者而来

Question

我正在使用Kafka Spring Integration发布和使用Kafka消息。我看到有效负载从生产者正确地传递到消费者，但是头信息在某个地方被覆盖了。

@ServiceActivator(inputChannel = "fromKafka")
public void processMessage(Message<?> message) throws InterruptedException,
        ExecutionException {
    try {
            System.out.println("Headers :" + message.getHeaders().toString());
        }
    } catch (Exception e) {
        e.printStackTrace();
    }
}

我得到以下标题：

Headers :{timestamp=1440013920609, id=f8c645f7-677b-ec32-dad0-a7b79082ef81}

我在生产者端构建消息，如下所示：

Message<FeelDBMessage> message = MessageBuilder
                .withPayload(samplePayloadObj)
                .setHeader(KafkaHeaders.MESSAGE_KEY, "key")
                .setHeader(KafkaHeaders.TOPIC, "sampleTopic").build();

        // publish the message
        publisher.publishMessage(message);

下面是生产者的标题信息：

 headers={timestamp=1440013914085, id=c4159c1c-2c67-634b-ef8d-3fb026b1172e, kafka_messageKey=key, kafka_topic=sampleTopic}

知道为什么头会被不同的值覆盖吗？

Answer 1

因为默认情况下Framework使用不可变的GenericMessage。

对现有消息的任何操作（例如MessageBuilder.WithPayload)都将生成一个新的GenericMessage实例。

另一方面，Kafka不支持任何头抽象，如JMS或AMQP。这就是为什么KafkaProducerMessageHandler只在向Kafka发布消息时这样做：

this.kafkaProducerContext.send(topic, partitionId, messageKey, message.getPayload());

正如您所看到的，它根本不发送标头。因此，另一方（消费者）只是将主题中的消息作为有效负载处理，并将一些系统选项作为头处理，如主题、分区、messagekey。

用两个词来说：我们不把头传给Kafka，因为它不支持他们。