问使用Spring Embedded Kafka测试@KafkaListener
EN

Stack Overflow用户

提问于 2018-05-01 20:48:22

回答 2查看 19.8K关注 0票数 20

我正在尝试为我正在使用Spring Boot2.x开发的Kafka侦听器编写单元测试。作为一个单元测试，我不想启动一个完整的Kafka服务器，一个Zookeeper的实例。因此，我决定使用Spring嵌入式Kafka。

我的听众的定义是非常基本的。

@Component
public class Listener {
    private final CountDownLatch latch;

    @Autowired
    public Listener(CountDownLatch latch) {
        this.latch = latch;
    }

    @KafkaListener(topics = "sample-topic")
    public void listen(String message) {
        latch.countDown();
    }
}

此外，在接收到消息后验证latch计数器是否等于零的测试也非常简单。

@RunWith(SpringRunner.class)
@SpringBootTest
@DirtiesContext
@EmbeddedKafka(topics = { "sample-topic" })
@TestPropertySource(properties = { "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}" })
public class ListenerTest {

    @Autowired
    private KafkaEmbedded embeddedKafka;

    @Autowired
    private CountDownLatch latch;

    private KafkaTemplate<Integer, String> producer;

    @Before
    public void setUp() {
        this.producer = buildKafkaTemplate();
        this.producer.setDefaultTopic("sample-topic");
    }

    private KafkaTemplate<Integer, String> buildKafkaTemplate() {
        Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
        ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
        return new KafkaTemplate<>(pf);
    }

    @Test
    public void listenerShouldConsumeMessages() throws InterruptedException {
        // Given
        producer.sendDefault(1, "Hello world");
        // Then
        assertThat(latch.await(10L, TimeUnit.SECONDS)).isTrue();
    }
}

不幸的是，测试失败了，我不明白为什么。是否可以使用KafkaEmbedded的实例来测试用注释@KafkaListener标记的方法

所有代码都在我的GitHub存储库kafka-listener中共享。

感谢所有人。

spring-kafka

spring-boot-test

java

spring-boot

apache-kafka

回答 2

Stack Overflow用户

回答已采纳

发布于 2018-05-01 21:17:59

您可能是在为使用者分配主题/分区之前发送消息。设置属性...

spring:
  kafka:
    consumer:
      auto-offset-reset: earliest

...it默认为latest。

这就像在控制台使用者中使用--from-beginning一样。

编辑

哦；你没有使用boot的属性。

添加

props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

EDIT2

顺便说一句，您可能还应该对template.send() ( Future<>)的结果执行get(10L, TimeUnit.SECONDS)，以断言发送成功。

EDIT3

要覆盖仅用于测试的偏移量重置，您可以执行与对代理地址相同的操作：

@Value("${spring.kafka.consumer.auto-offset-reset:latest}")
private String reset;

...

    props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, this.reset);

和

@TestPropertySource(properties = { "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
        "spring.kafka.consumer.auto-offset-reset=earliest"})

但是，请记住，此属性仅在组第一次使用时应用。要在每次应用程序启动时始终在末尾启动，您必须在启动期间一直搜索到末尾。

此外，我建议将enable.auto.commit设置为false，以便容器负责提交偏移量，而不是仅依赖于消费者客户端按时间计划执行该操作。

票数 14