在Spring Kafka Consumer Unit Test中没有调用MessageListner的onMessage

Posted

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了在Spring Kafka Consumer Unit Test中没有调用MessageListner的onMessage相关的知识,希望对你有一定的参考价值。

我正在编写Kafka Consumer Unit Test,需要模拟我的KafkaConsumer的服务,以独立测试Kafka Consumer。但是,不会调用服务的mockObject,而是Spring正在创建原始的Service类对象并调用它。因此,我的模拟类对象没有被调用。

卡夫卡消费者:

@Slf4j
@Component
@RequiredArgsConstructor (onConstructor = @__(@Autowired))
public class KafkaEventConsumer {

  private final MyService requestService;

  @KafkaListener (topics = "${kafka.topic:topic-name}")
  public void receive(@Payload String message) throws Exception {
    try {
      LOGGER.debug("Received message:{} ", message);
      ObjectMapper mapper = new ObjectMapper();
      ForecastRequest forecastRequest = mapper.readValue(message, ForecastRequest.class);

      JobDetail jobDetail = requestForecastService.refreshForecasts(forecastRequest);
      if (jobDetail.getJobStatus() != JobStatus.complete) {
        LOGGER.error("Failed to Refresh Forecast for ProgramId-{}, JobId-{}, JobStatus-{}",
            forecastRequest.getProgramId(), jobDetail.getJobId(), jobDetail.getJobStatus());
        throw new Exception("Internal Server Error");
      }
    } catch (Exception e) {
      LOGGER.error("Failed to Refresh Forecast for Forecast Request {}", message, e);
      throw e;
    }
  }
}

卡夫卡消费者测试:

@RunWith (SpringRunner.class)
@ActiveProfiles ("kafkatest")
@SpringBootTest (classes = ForecastEventConsumerApplication.class)
@DirtiesContext
public class KafkaEventConsumerTest {

  private static String TOPIC = "topic-name";

  @Mock
  private MyServiceImpl myServiceMock;

  @InjectMocks
  private KafkaEventConsumer kafkaEventConsumer;

  private KafkaTemplate<String, String> template;

  @Autowired
  private KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

  @ClassRule
  public static final KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true,3, TOPIC);


  @Before
  public void setUp() throws Exception {
    kafkaEventConsumer = new KafkaEventConsumer(myServiceMock);


    // set up the Kafka producer properties
    Map<String, Object> senderProperties = KafkaTestUtils.senderProps(embeddedKafka.getBrokersAsString());

    // create a Kafka producer factory
    ProducerFactory<String, String> producerFactory = new DefaultKafkaProducerFactory<String, String>(senderProperties);

    // create a Kafka template
    template = new KafkaTemplate<>(producerFactory);
    // set the default topic to send to
    template.setDefaultTopic(TOPIC);

    // wait until the partitions are assigned
    for (MessageListenerContainer messageListenerContainer : kafkaListenerEndpointRegistry.getListenerContainers()) {
      messageListenerContainer.setupMessageListener(new MessageListener<String, String>() {
        @Override
        public void onMessage(ConsumerRecord<String, String> record) {
          try {
            kafkaEventConsumer.receive(record.value());
          } catch (Exception e) {
            e.printStackTrace();
          }
        }
      });
      ContainerTestUtils.waitForAssignment(messageListenerContainer, embeddedKafka.getPartitionsPerTopic());
    }


  }

  @AfterClass
  public static void tearDown() throws Exception {
    embeddedKafka.destroy();
  }


  @Test
  public void testReceive() throws Exception {
    String forecastRequestMessage = "{"programId":100011770}";
    ForecastRequest forecastRequest = ForecastRequest.builder().programId(100011770L).build();
    JobDetail jobDetail = JobDetail.builder().jobStatus(JobStatus.complete).build();
    Mockito.when(forecastServiceMock.refreshForecasts(Matchers.any())).thenReturn(jobDetail);
    template.sendDefault(forecastRequestMessage);
    Thread.sleep(2000L);
    // validate something
  }

}

问题是,在上面的@Test方法而不是调用MyService的模拟版本,它调用原始的MyService实现。此外,在调试我的代码时,我发现被覆盖的onMessage()也没有被调用。请帮助我找到我在这里做错了什么。

答案

在调用他们的stop()之前,你必须MessageListenerContainer所有setupMessageListener()s。然后你需要回到start()让他们接一个新的听众:

protected void doStart() {
    ...
    Object messageListener = containerProperties.getMessageListener();
    Assert.state(messageListener != null, "A MessageListener is required");

无论如何,听起来你真的想要只嘲笑你注入真正的MyServiceKafkaEventConsumer。那么,如何考虑使用这样:

@MockBean
private MyServiceImpl myServiceMock;

你不需要在你的@Before做任何事情,也不需要在@InjectMocks

KafkaEmbedded可以将其主机/端口(或代理)属性暴露给预期的Spring Boot常规配置属性,如下所示:

@BeforeClass
public static void setup() {
    System.setProperty("spring.kafka.bootstrap-servers", kafkaEmbedded.getBrokersAsString());
}

https://docs.spring.io/spring-boot/docs/2.0.0.RELEASE/reference/htmlsingle/#boot-features-testing-spring-boot-applications-mocking-beans

以上是关于在Spring Kafka Consumer Unit Test中没有调用MessageListner的onMessage的主要内容,如果未能解决你的问题,请参考以下文章

总结kafka的consumer消费能力很低的情况下的处理方案

Kafka Consumer 指标在从 Spring Boot 2.2.2 升级到 2.3.0 后消失

在Spring Kafka Consumer Unit Test中没有调用MessageListner的onMessage

Spring Integration Kafka Consumer Listener 不接收消息

springboot kafka集成(实现producer和consumer)

从 Spring Boot 2.1.6 升级到 2.2.2 时,Prometheus 不导出 Kafka Consumer 数据