Test de integrare pt Kafka Consumer

@EmbeddedKafka
@SpringBootTest(properties = "spring.kafka.consumer.bootstrap-servers=${spring.embedded.kafka.brokers}")
public class ProductCreatedEventHandlerTest {
    @MockBean
    ProcessedEventRepository processedEventRepository;

    @Autowired
    KafkaTemplate<String, Object> kafkaTemplate;

    @SpyBean // true object, its methods can be intercepted
    ProductCreatedEventHandler productCreatedEventHandler;

    @Test
    public void testHandle() throws Exception {
        // Arrange
        ProductCreatedEvent event = new ProductCreatedEvent(UUID.randomUUID().toString(), "iPhone SE", BigDecimal.valueOf(24.5), 12);
        ProducerRecord<String, Object> record = new ProducerRecord<>(PRODUCT_CREATED_EVT_TOPIC, event.getProductId(), event);
        String messageId = UUID.randomUUID().toString();
        record.headers().add("messageId", messageId.getBytes());
        record.headers().add(KafkaHeaders.RECEIVED_KEY, event.getProductId().getBytes());

        ProcessedEventEntity processedEventEntity = new ProcessedEventEntity();
        when(processedEventRepository.findByMessageId(any())).thenReturn(processedEventEntity);

        // Act
        kafkaTemplate.send(record).get();

        // Assert
        ArgumentCaptor<String> messageIdCaptor = ArgumentCaptor.forClass(String.class);
        ArgumentCaptor<String> messageKeyCaptor = ArgumentCaptor.forClass(String.class);
        ArgumentCaptor<ProductCreatedEvent> eventCaptor = ArgumentCaptor.forClass(ProductCreatedEvent.class);
        verify(productCreatedEventHandler, timeout(5000).times(1))
                .handle(eventCaptor.capture(), messageIdCaptor.capture(), messageKeyCaptor.capture());
        Assertions.assertEquals(messageId, messageIdCaptor.getValue());
        Assertions.assertEquals(event.getProductId(), messageKeyCaptor.getValue());
        Assertions.assertEquals(event.getProductId(), eventCaptor.getValue().getProductId());
    }
}