Создать собственный DefaultKafkaHeaderMapper

Question

Когда я отправляю запись в тему kafka, потребитель получает «nativeHeaders» с некоторым ненужным заголовком (который HeaderMethodArgumentResolver не может даже привести к Map).

Я ищу какой-то способ переопределить метод HeaderMethodArgumentResolver "getNativeHeaders", чтобы исключить этот заголовок мусора и не знаю, как предоставить этот подкласс для пружины.

Существует оригинальный метод изorg.springframework.messaging.handler.annotation.support.HeaderMethodArgumentResolver:

private Map<String, List<String>> getNativeHeaders(Message<?> message) {
   return (Map)message.getHeaders().get("nativeHeaders");
  }

Где этот вызов:

message.getHeaders().get("nativeHeaders");

возвращает это: https://ibb.co/qrvMNMk (как вы видите, естьдополнительное поле "headerValue" кроме ключа-значения, которое предотвращает приведение)

Отправить запись с помощью kafkaTemplate следующим образом:

kafkaTemplate.send(new ProducerRecord<String, TempContractEntity>(topics.getSubmit(), tempContractEntity));

Потребитель получает сообщения от @KafkaListener аннотация:

@KafkaListener(topics = "#{settingsService.getTopics()}")
public void processMessage(OrchestratorRequestImpl orchestratorRequest,
                           @Header(KafkaHeaders.RECEIVED_TOPIC) String topicName) throws Throwable{//...}

Как правило, я хочу найти способ предварительной обработки заголовков кафки

Answer 1

NonTrustedHeaderType указывает, что что-то sent сообщение с этим заголовком и его классом не является доверенным. Этого не произойдет с типом отправки, который вы показываете - там нет Message<?>, поэтому что-то отсутствует на картинке в вашем вопросе.

Одна вещь, которую вы можете сделать, это добавить ConsumerInterceptor к конфигурации потребителя и отсеять нежелательный заголовок в методе onConsume().

Но вы действительно должны выяснить, кто его отправляет.