Test Kafka consumer with Testcontainers and Mockito couldn't invoke consumer in SpringBoot

I am trying to setup a kafka consumer and test it with Testcontainers and Mickito. It looks like the producer works fine but the consumer has not been triggered and always gives me:

Wanted but not invoked:
helloKafkaService.handleMessage(
<Capturing argument>
);
-> at com.my.kafka.HelloKafkaListenerTestIT.helloKafkaListenerTest
Actually, there were zero interactions with this mock.

Here is my service interface HelloKafkaService.java:

@Service
public interface HelloKafkaService {
public void handleMessage(ExampleEvent exampleEvent);
}

My consumer: HelloKafkaListener.java

@Component
public class HelloKafkaListener {
private static final Logger log = LoggerFactory.getLogger(HelloKafkaListener.class);
private final HelloKafkaService helloKafkaService;
public HelloKafkaListener(HelloKafkaService helloKafkaService) {
this.helloKafkaService = helloKafkaService;
}
@KafkaListener(
topics = "my-topic",
groupId = "my-topic:HelloKafkaListener")
public void process(ExampleEvent event) {
this.helloKafkaService.handleMessage(event);
log.info("Processing event: " + event.getExampleField());
}
}

My test HelloKafkaListenerTestIT.java:

@SpringBootTest
@Testcontainers
public class HelloKafkaListenerTestIT {
@Container
public static KafkaContainer kafkaContainer = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.0.5"));
@Mock
private HelloKafkaService helloKafkaService;
@Autowired
private KafkaTemplate<String, ExampleEvent> kafkaTemplate;
@DynamicPropertySource
static void kafkaProperties(DynamicPropertyRegistry registry) {
registry.add(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer::getBootstrapServers);
registry.add(ProducerConfig.CLIENT_ID_CONFIG, () -> "test-id");
registry.add(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, () -> StringSerializer.class.getName());
registry.add(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, () -> KafkaAvroSerializer.class.getName());
}
@BeforeEach
public void setUp() {
Serializer keySerializer;
Serializer valueSerializer;
var avroConfig = Map.of(
KafkaAvroSerializerConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081",
KafkaAvroSerializerConfig.AUTO_REGISTER_SCHEMAS, true,
KafkaAvroSerializerConfig.USE_LATEST_VERSION, true
);
keySerializer = new StringSerializer();
valueSerializer = new SpecificAvroSerializer<ExampleEvent>();
valueSerializer.configure(avroConfig, false);
Map<String, Object> config = new HashMap<>();
config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaContainer.getBootstrapServers());
config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
config.put(ProducerConfig.CLIENT_ID_CONFIG, "test-id");
config.put("schema.registry.url", "http://localhost:8081");
DefaultKafkaProducerFactory<String, ExampleEvent> producerFactory =
new DefaultKafkaProducerFactory<>(config, keySerializer, valueSerializer);
this.kafkaTemplate = new KafkaTemplate<>(producerFactory);
}
static {
kafkaContainer.start();
}
@Test
public void helloKafkaListenerTest() {
ArgumentCaptor<ExampleEvent> captor = ArgumentCaptor.forClass(ExampleEvent.class);
ExampleEvent exampleEvent = new ExampleEvent("Hello World!");
// Confirmed this.kafkaTemplate.send works well
this.kafkaTemplate.send("my-topic", "key", exampleEvent);
// Throw the error here:
verify(helloKafkaService, timeout(5000)).handleMessage(captor.capture());
assertNotNull(captor.getValue());
assertEquals("Hello World!", captor.getValue().getExampleField());
}
@AfterAll
static void tearDown() {
kafkaContainer.stop();
}

event.avsc:

{
"namespace": "my.namespace",
"type": "record",
"name": "ExampleEvent",
"doc": "A sample event",
"fields": [
{
"name": "exampleField",
"type": "string"
}
]
}

Why can the consumer not be triggered, if using @Mock and verify? Is there a better idea to test this?

Because @Mock just create a mocked instance but this instance will not be managed by the spring context. So HelloKafkaService inside HelloKafkaListener is still the actual instance but not the mocked one.

You have to use @MockBean. The difference is that it will create a mocked instance and also replace the actual bean instance in the spring context with it.

So change to the following should fix the problem :

@SpringBootTest
@Testcontainers
public class HelloKafkaListenerTestIT {
  
  @MockBean
  private HelloKafkaService helloKafkaService;

}