DirtiesContext does not close Kafka consumer and producer

Question

I'm trying to test a KafkaStreams application.Using JUnit5 and EmbededKafka. When executing any TestCase, I see that the logs are bombarded with the following messages

[Producer clientId=myTask-227e2e90-212b-4663-bd17-2d307018c81a-StreamThread-1-producer] Connection to node 0 (localhost/127.0.0.1:61269) could not be established. Broker may not be available.
[Consumer clientId=myTask-227e2e90-212b-4663-bd17-2d307018c81a-StreamThread-3-consumer, groupId=my-task] Connection to node 0 (localhost/127.0.0.1:61269) could not be established. Broker may not be available.

I have also added

@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)

Class Level configurations

@SpringBootTest(webEnvironment = SpringBootTest.WebEnvironment.NONE)
@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD)
@TestPropertySource("/application-test.properties")
@EmbeddedKafka(topics = { "${kafka.topic.}"
    },
    partitions = 1, controlledShutdown = true
)
@ExtendWith(MockitoExtension.class)
public class KStreamTest{

    @Autowired
 protected EmbeddedKafkaBroker embeddedKafka;
  
  //TestMethod

 }

Can you suggest anything I can try or if I'm missing something?

Creationism

Just a comment: This will make your tests crawl. Each startup of kafka takes multiple seconds and they will happen for each test. I don't know about your app, but there are some approaches to write tests that avoid having to reload an app for each test. Just sharing this as I recently saw a project where this was done extensively and the test suite took ~40mins to run. We avoided reloading kafk (or using DirtiesContext) in another project and the test suite took less than 4 minutes to run and afterwards we could see the data in kafka and the DB if we needed. – Priddy

@Priddy Could you share some details on how you replicated the DirtiesContext behaviour without using the annotation? – Creationism

I didn't replicate the behaviour, but designed the configuration classes so it wouldn't be required for the tests. If you need the annotation it means that there's a side effect of a test that can affect other tests, the solution is to address that side effect so the tests can run independently. In the case of Kafka it was to find for a message that matched the expectation rather than expect 1 message with a given payload. – Priddy

Answer 1

I had a similar problem with our producers between multiple test cases. The problem was, that the ProdcuerFactory used by the KafkaTemplate was not registered as Bean and thus not destroyed by the @DirtiesContext-Annotation.

So your Bean definition should look like this:

@Configuration
public class KafkaProducerConfig {

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        Map<String, Object> configProps = new HashMap<>();
        // ...
        return new DefaultKafkaProducerFactory<>(configProps);
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }
}

Falconet