我正在编写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");
无论如何，听起来你真的想要只嘲笑你注入真正的MyService的KafkaEventConsumer。那么，如何考虑使用这样：
@MockBean
private MyServiceImpl myServiceMock;
你不需要在你的@Before做任何事情，也不需要在@InjectMocks。
KafkaEmbedded可以将其主机/端口（或代理）属性暴露给预期的Spring Boot常规配置属性，如下所示：
@BeforeClass
public static void setup() {
System.setProperty("spring.kafka.bootstrap-servers", kafkaEmbedded.getBrokersAsString());
}