需要帮助 |单元测试 | Kafka 自定义 ConcurrentKafkaListenerContainerFactory 和自定义记录 FilterStrategy

Question

尝试使用自定义 RecordFilterStrategy 对 ConcurrentKafkaListenerContainerFactory 进行单元测试，但无法找到测试过滤器策略的最佳方法。

  class ConsumerConfiguration {

  @Bean
  public ConcurrentKafkaListenerContainerFactory<String, Message> messageListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, Message> factory =
        new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setRecordFilterStrategy(consumerRecord -> consumerRecord.value().getType().contains("MYFILTER"));
    return factory;
  }

}

单元测试

  @Mock
  KafkaProperties kafkaProperties;
  ConsumerConfiguration configuration;
  @Test
  void testMessageListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, Message> actual =
        configuration.messageListenerContainerFactory();
    assertEquals(kafkaProperties.buildConsumerProperties(),
        actual.getContainerProperties().getKafkaConsumerProperties());
  }

虽然这段代码对单元测试很有用，但是没有提供自定义 RecordFilterStrategy 的代码覆盖，即 lambda 函数，所以如果有人已经为 Kafka Listener Container Factory 完成了单元测试/代码覆盖，那么需要帮助，什么是最好的方法处理。

Answer 1

对于工厂配置的纯单元测试，过滤器在哪里

record -> record.value().contains("foo")

你可以做这样的事情......

@SpringBootTest
class So63239177ApplicationTests {

    @Autowired
    ConcurrentKafkaListenerContainerFactory<String, String> factory;

    private int called;

    @Test
    void testFilter() throws Exception {
        MethodKafkaListenerEndpoint<String, String> endpoint = new MethodKafkaListenerEndpoint<>();
        endpoint.setBean(this);
        endpoint.setMethod(getClass().getDeclaredMethod("listen", ConsumerRecord.class));
        endpoint.setMessageHandlerMethodFactory(new DefaultMessageHandlerMethodFactory());
        endpoint.setTopics("foo");
        ConcurrentMessageListenerContainer<String, String> container = factory.createListenerContainer(endpoint);
        @SuppressWarnings("unchecked")
        MessageListener<String, String> messageListener =
                (MessageListener<String, String>) container.getContainerProperties().getMessageListener();
        messageListener.onMessage(new ConsumerRecord<>("foo", 0, 0L, null, "foo"));
        assertThat(this.called).isEqualTo(0);
        messageListener.onMessage(new ConsumerRecord<>("foo", 0, 0L, null, "bar"));
        assertThat(this.called).isEqualTo(1);
    }

    public void listen(ConsumerRecord<String, String> in) {
        this.called++;
    }

}

编辑

这是一种稍微简单（也许更清洁）的方法......

@SpringBootTest
class So632391771ApplicationTests {

    @Autowired
    KafkaListenerAnnotationBeanPostProcessor<?, ?> bpp;

    @Autowired
    KafkaListenerEndpointRegistry registry;

    @Test
    void testFilter() throws Exception {
        TestListener listener = new TestListener();
        this.bpp.postProcessAfterInitialization(listener, "test");
        ContainerProperties containerProperties = registry.getListenerContainer("test")
                .getContainerProperties();
        assertThat(containerProperties.getGroupId()).isEqualTo("test");
        @SuppressWarnings("unchecked")
        MessageListener<String, String> messageListener =
                    (MessageListener<String, String>) containerProperties.getMessageListener();
        messageListener.onMessage(new ConsumerRecord<>("foo", 0, 0L, null, "foo"));
        assertThat(listener.called).isEqualTo(0);
        messageListener.onMessage(new ConsumerRecord<>("foo", 0, 0L, null, "bar"));
        assertThat(listener.called).isEqualTo(1);
    }

}

class TestListener { // NOTE: Not a @Bean

    int called;

    @KafkaListener(id = "test", topics = "foo", autoStartup = "false")
    public void listen(String in) {
        this.called++;
    }

}