Spring KafkaEmbedded - проблема потребления сообщений

Question

У меня проблема с использованием KafkaEmbedded от https://mvnrepository.com/artifact/org.springframework.kafka/spring-kafka-test/2.1.10.RELEASE

Я использую KafkaEmbedded для создания брокера Kafka для тестирования конвейеров производителя / потребителя.Эти производители / потребители являются стандартными клиентами от kafka-клиентов.Я не использую клиенты Spring Kafka.

Все работает, код работает нормально, но я должен использовать метод consumeFromEmbeddedTopics() из KafkaEmbedded, чтобы потребитель работал.Если я не буду использовать этот метод, потребитель не получит никаких сообщений.

У меня есть две проблемы с этим методом: во-первых, ему нужен KafkaConsumer в качестве параметра (и я не хочувыставить его в классе) и вызов этого метода дает ConcurrentModificationException, когда объект вызывает опрос с использованием @Scheduled.

Я использую свойство auto.offset.reset, так что это совсем другое.

У меня вопрос: как правильно использовать записи из KafkaEmbedded без вызова этих consumeFromEmbeddedTopics() методов?

Answer 1

В этом методе нет ничего особенного, он просто подписывает потребителя на тему (и) и опрашивает его.

Нет причины, по которой вы не можете сделать то же самое с вашим потребителем.