Блог

Я использую Spring Kafka с @EmbeddedKafka аннотацией для запуска встроенного экземпляра kafka.

Зависимость:

 <dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka-test</artifactId>
  <scope>test</scope>
</dependency>
  

И я настроил свое приложение для подключения к нему:

 spring:
  kafka:
    bootstrap-servers: ${spring.embedded.kafka.brokers}
  

Я успешно отправляю сообщения и получаю сообщения.

Моя проблема в том, что сообщения где-то сохраняются и могут появляться при продолжающихся тестовых запусках. И журналы печатают эти:

 Successfully joined group with generation 117
  

Высокий номер поколения указывает на то, что Kafka сохраняет информацию между повторными запусками тестов, которая мне не нужна.

Как мне полностью очистить встроенную Kafka и начать заново?

Редактировать:

Проблема, с которой я столкнулся, заключалась в том, что я неправильно использовал профили Spring. У меня была пользовательская аннотация аннотации с моей встроенной конфигурацией Kafka. Проблема заключалась в том, что я устанавливал @ActiveProfiles("kafka") в составной аннотации, а также устанавливал @ActiveProfiles("dev") в реальном тестовом классе. Я обновил составную аннотацию к этому:

 @Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
// @ActiveProfiles("kafka") REMOVED
@TestPropertySource(properties = "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}")
@EmbeddedKafka(/* config properties */)
public @interface CustomEmbeddedKafka{
}
  

Вам необходимо использовать @DirtiesContext для очистки контекста приложения между тестами. Вместе с этим встроенный брокер Kafka также будет уничтожен.

Если проблема заключается в методах одного тестового класса, то вам нужно переосмыслить логику, чтобы не мешать одним и тем же темам в разных методах.

Вы также можете рассмотреть возможность использования @DirtiesContext(classMode = DirtiesContext.ClassMode.BEFORE_EACH_TEST_METHOD) , но это приведет к некоторому снижению производительности, поскольку контекст приложения вместе со встроенной Kafka будет воссоздаваться перед каждым методом тестирования.

Обновить

Попробуйте это предложение вместо application.properties модификации:

 static {
    System.setProperty(EmbeddedKafkaBroker.BROKER_LIST_PROPERTY,
            "spring.kafka.bootstrap-servers");
}
  

См. https://docs.spring.io/spring-kafka/docs/2.2.5.RELEASE/reference/#kafka-testing-embeddedkafka-annotation