如何测试是否正在调用带有@KafkaListener 的方法

Question

我真的很难写一个测试来检查当消息发送到它的指定主题时我的 Kafka 消费者是否被正确调用。

我的消费者：

@Service
@Slf4j
@AllArgsConstructor(onConstructor = @__(@Autowired))
public class ProcessingConsumer {

  private AppService appService;

  @KafkaListener(
      topics = "${topic}",
      containerFactory = "processingConsumerContainerFactory")
  public void listen(ConsumerRecord<Key, Value> message, Acknowledgment ack) {
    try {
      appService.processMessage(message);
      ack.acknowledge();
    } catch (Throwable t) {
      log.error("error while processing message!", t);
    }
  }
}

我的消费者配置：

@EnableKafka
@Configuration
public class ProcessingCosumerConfig {

  @Value("${spring.kafka.schema-registry-url}")
  private String schemaRegistryUrl;

  private KafkaProperties props;

  public ProcessingCosumerConfig(KafkaProperties kafkaProperties) {
    this.props = kafkaProperties;
  }

  public Map<String, Object> deserializerConfigs() {
    Map<String, Object> props = new HashMap<>();
    props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true);
    props.put(KafkaAvroDeserializerConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);

    return props;
  }

  private KafkaAvroDeserializer getKafkaAvroDeserializer(Boolean isKey) {
    KafkaAvroDeserializer kafkaAvroDeserializer = new KafkaAvroDeserializer();
    kafkaAvroDeserializer.configure(deserializerConfigs(), isKey);
    return kafkaAvroDeserializer;
  }

  private DefaultKafkaConsumerFactory consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(
        props.buildConsumerProperties(),
        getKafkaAvroDeserializer(true),
        getKafkaAvroDeserializer(false));
  }

  @Bean(name = "processingConsumerContainerFactory")
  public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Key, Value>>
      kafkaListenerContainerFactory() {

    ConcurrentKafkaListenerContainerFactory<Key, Value>
        factory = new ConcurrentKafkaListenerContainerFactory<>();

    factory.setConsumerFactory(consumerFactory());
    factory.getContainerProperties().setAckOnError(false);
    factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
    factory.setErrorHandler(new SeekToCurrentErrorHandler());
    return factory;
  }
}

最后，我的（想要的）测试：

@DirtiesContext
public class ProcessingConsumerTest extends BaseIntegrationTest{

  @Autowired private ProcessingProducerFixture processingProducer;
  @Autowired private ProcessingConsumer processingConsumer;
  @org.springframework.beans.factory.annotation.Value("${topic}")
  String topic;

  @Test
  public void consumer_shouldConsumeMessages_whenMessagesAreSent() throws Exception{
    Thread.sleep(1000);
    ProducerRecord<Key, Value> message = new ProducerRecord<>(topic, new Key("b"), new Value("a", "b", "c", "d"));
    processingProducer.send(message);
  }
}

这就是我目前所拥有的一切。我已经尝试检查这种方法是否使用调试手动到达消费者，甚至只是在那里放置简单的打印件，但执行似乎根本没有到达那里。此外，如果我的测试可以以某种方式正确调用它，我不知道如何在实际测试中实际断言它。

Answer 1

将模拟 AppService 注入侦听器并验证其 processMessage() 已被调用。

如何测试是否正在调用带有@KafkaListener 的方法

How to test if method with @KafkaListener is being called

java

apache-kafka

spring-kafka