Потребляйте сообщение Kafka с пользовательским заголовком ⇐ JAVA

[Anonymous]

Я пытаюсь создать потребителя Kafka с Spring Cloud Stream, чтобы прослушать сообщение Kafka, созданное вне любого весеннего контекста, с пользовательским заголовком ( anpormationtype < /strong>). < /p>

Я использую Spring Boot 1.5.x /Spring Egdware.sr5 и 1.1.1 версии Kafka и Kafka_2.11. < /p>

Мой класс слушателя содержит этот метод < /p>

Код: Выделить всё

@StreamListener(value = "dataset-changed", condition = "headers['operationType']=='UPDATE'")
public void onEvent(@Payload DatasetChangedMessage payload) {
// my code should be execute only if the header operationType == UPDATE
}
< /code>

Конфигурация облачного потока пружины - < /p>

spring.cloud.stream:
bindings:
dataset-changed:
group: preparation
content-type: application/json
destination: dataset-changed
consumer:
headerMode: raw
configuration:
key.deserializer: org.apache.kafka.common.serialization.ByteArrayDeserializer
value.deserializer: org.apache.kafka.common.serialization.StringDeserializer
< /code>

Производитель-это простой Java Doc с Kafka-client: 1.1.1 Библиотека < /p>

Properties producerConfig = new Properties();
producerConfig.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
producerConfig.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArraySerializer");
producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

KafkaProducer producer = new KafkaProducer(producerConfig);

// Headers (to condition the kafka listener)
final List headers = new ArrayList();
headers.add(new RecordHeader("operationType", "UPDATE".getBytes()));

ProducerRecord record =
new ProducerRecord("dataset-changed", 0, "111".getBytes(), getJsonPayload(), headers);
Future future = producer.send(record);
future.get();

producer.close();
< /code>

Когда я создаю сообщение Kafka, у меня есть такие предупреждения < /p>

2019-03-15 14:48:32.103  WARN [tdp-preparation,1e24b9764ef9bb14,1e24b9764ef9bb14,false] 34760 --- [           -L-1] .DispatchingStreamListenerMessageHandler : Cannot find a @StreamListener matching for message with id: ea27a446-69da-7b8d-1b94-50b46a40dfde

, тогда как заголовок OperationType присутствует



Подробнее здесь: https://stackoverflow.com/questions/551 ... tom-header