覆盖在 Spring Kafka Consumer Unit Test 中未调用的消息列表器的消息

Question

我正在编写Kafka消费者单元测试，需要模拟我的KafkaConsumer的服务以独立测试Kafka消费者。但是，mockObject of Service没有被调用，而是Spring正在创建原始的服务类对象并调用它。因此，我的模拟类对象没有被调用。

卡

夫卡消费者 ：

@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()也没有被调用。请帮助我找到我在这里做错了什么。

Answer 1

在调用其setupMessageListener()之前，您必须stop()所有MessageListenerContainer。然后你需要把它们start()回来，让他们挑选一个新的听众：

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

无论如何，听起来你真的只想嘲笑你注入真实KafkaEventConsumer MyService。那么，如何考虑像这样使用它：

@MockBean
private MyServiceImpl myServiceMock;

而且您无需在@Before中执行任何操作，也无需在@InjectMocks中执行任何操作.

KafkaEmbedded可以向预期的 Spring 引导常规配置属性公开其主机/端口(或代理(属性，如下所示：

@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