ReplyingKafkaTemplate / KafkaTemplate не отправляет / получает ключ

Question

У меня есть шаблон, который выглядит следующим образом:

@Autowired
private ReplyingKafkaTemplate<ItemId, MessageBDto, MessageBDto> xxx2ReplyingKafkaTemplate;

Мой метод отправки оболочки выглядит следующим образом:

public RequestReplyFuture<ItemId, MessageBDto, MessageBDto> sendAndReceiveMessageB(MessageBDto message) {
    ProducerRecord<ItemId, MessageBDto> producerRecord = new ProducerRecord<>(KafkaTopicConfig.xxx2_TOPIC, new ItemId(message.getCount()), message);
    producerRecord.headers().add(new RecordHeader(KafkaHeaders.REPLY_TOPIC, KafkaTopicConfig.xxx2_REPLY_TOPIC.getBytes()));
    return this.xxx2ReplyingKafkaTemplate.sendAndReceive(producerRecord);
}

И это мой слушатель:

@SendTo
@KafkaListener(topics=KafkaTopicConfig.xxx2_TOPIC, containerFactory="xxx2ListenerContainerFactory")
public MessageBDto xxx2Listener(ConsumerRecord<ItemId, MessageBDto> message) {
    System.out.println("xxx2(value): " + message.value().getMessage() + ", " + message.value().getCount());
    message.value().setCount(message.value().getCount() * 2);
    return message.value();
}

Разве это не должно отправлять Key = ItemId, Value = MessageBDto и получать ключ в слушателе?

Кажется, что слушатель не получает ключ и / или кажется, что он другой Экземпляр MessageBDto.

Не понимаю ли я, как это должно работать?

РЕДАКТИРОВАТЬ:

БИНЫ ПРОИЗВОДИТЕЛЯ:

@Bean
public ProducerFactory<ItemId, MessageBDto> xxx2ProducerFactory() {
    return new DefaultKafkaProducerFactory<ItemId, MessageBDto>(super.producerConfigs(),
                                                                new JsonSerializer<ItemId>(),
                                                                new JsonSerializer<MessageBDto>());
}

@Bean
public ConsumerFactory<ItemId, MessageBDto> xxx2ConsumerFactory() {
    return new DefaultKafkaConsumerFactory<>(super.consumerConfigs(),
                                             trustingDeserializer(ItemId.class),
                                             trustingDeserializer(MessageBDto.class));
}

@Bean
public KafkaMessageListenerContainer<ItemId, MessageBDto> dtms2MessageListenerContainer() {
    return new KafkaMessageListenerContainer<>(xxx2ConsumerFactory(),
                                               new ContainerProperties(KafkaTopicConfig.xxx2_REPLY_TOPIC));
}

@Bean
public ReplyingKafkaTemplate<ItemId, MessageBDto, MessageBDto> xxx2ReplyingKafkaTemplate() {
    return new ReplyingKafkaTemplate<>(xxx2ProducerFactory(), xxx2MessageListenerContainer());
}

private <T> JsonDeserializer<T> trustingDeserializer(Class<T> targetType) {
    JsonDeserializer<T> deserializer = new JsonDeserializer<>(targetType);
    deserializer.addTrustedPackages("*");
    return deserializer;
}

ПОТРЕБИТЕЛЬСКИЕ БИНЫ:

@Bean
public KafkaTemplate<ItemId, MessageBDto> xxx2KafkaTemplate() {
    return new KafkaTemplate<>(xxx2ProducerFactory());
}

@Bean
public KafkaListenerContainerFactory<ItemId, MessageBDto> xxxListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<ItemId, MessageBDto> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(dtms2ConsumerFactory());
    factory.setReplyTemplate(xxx2KafkaTemplate());
    return factory;
}

Когда я смотрю в отладчике моего слушателя, он показывает, что ключ является пустым экземпляром MessageBDto ???

Версии:

• Apache Zookeeper 3.5.7
• Apache Кафка 2.12-2.4.0

• Пружинный ботинок 2.2.4

  <dependency>
      <groupId>org.springframework.kafka</groupId>
      <artifactId>spring-kafka</artifactId>
      <version>2.4.2.RELEASE</version> <!-- $NO-MVN-MAN-VER$ -->
  </dependency>
  <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka-streams</artifactId>
      <version>2.4.0</version> <!--$NO-MVN-MAN-VER$-->
  </dependency>
  <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka-clients</artifactId>
      <version>2.4.0</version>  <!--$NO-MVN-MAN-VER$-->
  </dependency>
  

Answer 1

Я не уверен, что происходит с вашим кодом. Вот рабочий пример ...

@SpringBootApplication
public class So60384112Application {


    private static final Logger LOG = LoggerFactory.getLogger(So60384112Application.class);


    public static void main(String[] args) {
        SpringApplication.run(So60384112Application.class, args).close();
    }

    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("so60384112").partitions(1).replicas(1).build();
    }

    @Bean
    public NewTopic replies() {
        return TopicBuilder.name("so60384112replies").partitions(1).replicas(1).build();
    }

    @KafkaListener(id = "so60384112", topics = "so60384112")
    @SendTo
    public Message<?> listen(ConsumerRecord<Foo, Bar> record) {
        LOG.info(record.key().toString() + ":" + record.value().toString());
        return MessageBuilder.withPayload(new Bar(record.value().getField().toUpperCase()))
                .setHeader(KafkaHeaders.MESSAGE_KEY, record.key())
                .setHeader(KafkaHeaders.CORRELATION_ID, record.headers().lastHeader(KafkaHeaders.CORRELATION_ID).value())
                .setHeader(KafkaHeaders.TOPIC, record.headers().lastHeader(KafkaHeaders.REPLY_TOPIC).value())
                .build();
    }

    @Bean
    public ReplyingKafkaTemplate<Foo, Bar, Bar> replyer(ProducerFactory<Foo, Bar> pf,
            ConcurrentKafkaListenerContainerFactory<Foo, Bar> containerFactory) {

        containerFactory.setReplyTemplate(kafkaTemplate(pf));
        ConcurrentMessageListenerContainer<Foo, Bar> container = containerFactory.createContainer("so60384112replies");
        container.getContainerProperties().setGroupId("so60384112replies");
        ReplyingKafkaTemplate<Foo, Bar, Bar> replyer = new ReplyingKafkaTemplate<>(pf, container);
        return replyer;
    }

    @Bean
    public KafkaTemplate<Foo, Bar> kafkaTemplate(ProducerFactory<Foo, Bar> pf) {
        return new KafkaTemplate<>(pf);
    }

    @Bean
    public ApplicationRunner runner(ReplyingKafkaTemplate<Foo, Bar, Bar> template) {
        return args -> {
            RequestReplyFuture<Foo, Bar, Bar> future =
                    template.sendAndReceive(new ProducerRecord<>("so60384112", 0, new Foo("foo"), new Bar("bar")));
            ConsumerRecord<Foo, Bar> record = future.get(10, TimeUnit.SECONDS);
            LOG.info(record.key().toString() + ":" + record.value().toString());
        };
    }

}

class Foo {

    private String field;

    public Foo() {
    }

    public Foo(String field) {
        this.field = field;
    }

    public String getField() {
        return this.field;
    }

    public void setField(String field) {
        this.field = field;
    }

    @Override
    public String toString() {
        return getClass().getSimpleName() + " [field=" + this.field + "]";
    }

}

class Bar  {

    private String field;

    public Bar() {
    }

    public Bar(String field) {
        this.field = field;
    }

    public String getField() {
        return this.field;
    }

    public void setField(String field) {
        this.field = field;
    }

    @Override
    public String toString() {
        return getClass().getSimpleName() + " [field=" + this.field + "]";
    }

}

spring.kafka.consumer.key-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.properties.spring.json.trusted.packages=*

spring.kafka.producer.key-serializer=org.springframework.kafka.support.serializer.JsonSerializer
spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

2020-02-24 18: 00: 47.904 INFO 16591 --- [o60384112-0- C -1] com .example.demo.So60384112Применение: Foo [field = foo]: Bar [field = bar]

2020-02-24 18: 00: 47.915 INFO 16591 --- [main] com.example.demo. So60384112Применение: Foo [field = foo]: Bar [field = BAR]

Чтобы вернуть ключ, вы должны вернуть Message<?>. К сожалению, вы должны установить заголовки для ответа topi c и корреляции тоже.