EmbeddedKafkaCluster missing?

Question

This blog entry https://www.confluent.io/blog/stream-processing-part-2-testing-your-streaming-application/ refers to the class EmbeddedKafkaCluster, which is supposed to be in the library kafka-streams-test-utils.

However, this class is missing in the library, e.g. org.apache.kafka/kafka-streams-test-utils/2.5.1.

I thought that I can use the source code from github https://github.com/a0x8o/kafka/blob/master/streams/src/test/java/org/apache/kafka/streams/integration/utils/EmbeddedKafkaCluster.java

But this source code refers to some classes, e.g. kafka.zk.EmbeddedZookeeper and kafka.utils.MockTime, that I assumed must be in the library like org.apache.kafka/kafka_2.13/2.5.1. Unfortunately, they are also missing.

What is the best way to configure a project to use the EmbeddedKafkaCluster in this case?

Thanks

Boris

Answer 1

Add the following dependencies:

    //build.gradle

    testCompile group: 'junit', name: 'junit', version: '4.13'
    testCompile group: 'org.hamcrest', name: 'hamcrest-junit', version: '2.0.0.0'

    compile group: 'org.apache.kafka', name: 'kafka_2.13', version: '2.7.0'
    testCompile group: 'org.apache.kafka', name: 'kafka_2.13', version: '2.7.0', classifier:'test'
    compile group: 'org.apache.kafka', name: 'kafka-streams', version: '2.7.0'
    testCompile group: 'org.apache.kafka', name: 'kafka-streams', version: '2.7.0', classifier: 'test'
    compile group: 'org.apache.kafka', name: 'kafka-streams-test-utils', version:  '2.7.0'
    testCompile group: 'org.apache.kafka', name: 'kafka-clients', version: '2.7.0', classifier: 'test'

If you’re using Maven, convert all dependencies based on this code:

//pom.xml
<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-streams</artifactId>
    <version>2.7.0</version>
    <scope>test</scope>
    <classifier>test</classifier>
</dependency>

And create the EmbeddedKafkaCluster in the following way (Kotlin example):

    @Test
    fun createEmbeddedKafkaClusterTest() {

        val NUM_BROKERS = 1
        val embeddedKafkaCluster = EmbeddedKafkaCluster(NUM_BROKERS)
        Assert.assertNotNull(embeddedKafkaCluster)

        embeddedKafkaCluster.start()

        embeddedKafkaCluster.createTopic("TestTopic")
    }