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

Question

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