When you start working with Quarkus, you discover the magic of dev services! If you have an application that use Kafka, a database, RabbitMQ, Redis, ... (full list here ), then if you don't have some properties set for dev and test profile like broker, url, ... Quarkus will start a dev service using Test Containers.

You get crazy, code as much as integration test as you can, it is so easy and finally push to your favourite CI/CD pipeline.

And now everything is red in your test job!!! Whyyyyy??? Well, your sys admin removed the dind from the runners... Yes using dind can create a security breach in your pipeline...

Of course they tell you that you can start a service (gitlab services for example) and plug your tests on it but you want it to work locally so having to start (maintain) a container, inject the correct urls, be sure to clean / setup the content properly,... no thanks.

We want to have something inside of the code.

We will see how to setup your Kafka cluster inside of your integration tests.

Create a Quarkus project

Let's go to code.quarkus.io, create a new project with those dependencies:

• SmallRye Reactive Messaging
• SmallRye Reactive Messaging - Kafka Connector

Now let's remove the classes generated (in main and test) and the META-INF

Create a new service KafkaService:

import io.smallrye.mutiny.Multi;
import java.time.Duration;
import javax.enterprise.context.ApplicationScoped;
import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.eclipse.microprofile.reactive.messaging.Message;
import org.eclipse.microprofile.reactive.messaging.Outgoing;
import org.jboss.logging.Logger;

@ApplicationScoped
public class KafkaService {

  public static final Logger LOGGER = Logger.getLogger(KafkaService.class);

  @Outgoing("demo-kafka-out")
  public Multi<Message<String>> produceData() {
    return Multi.createFrom().ticks().every(Duration.ofSeconds(1))
        .map(tick -> Message.of("SmallRye Hello " + tick));
  }

  @Incoming("demo-kafka-in")
  public void consumeData(String message) {
    LOGGER.infov("Message received {0}", message);
  }
}

We have a simple example with one producer that will push every second a message and one consumer that will log this message.

We have to set the properties:

mp.messaging.outgoing.demo-kafka-out.connector=smallrye-kafka
mp.messaging.incoming.demo-kafka-in.connector=smallrye-kafka
mp.messaging.outgoing.demo-kafka-out.topic=demo-kafka
mp.messaging.incoming.demo-kafka-in.topic=demo-kafka
mp.messaging.outgoing.demo-kafka-out.value.serializer=org.apache.kafka.common.serialization.StringSerializer
mp.messaging.incoming.demo-kafka-in.value.deserializer=org.apache.kafka.common.serialization.StringDeserializer

Create a new empty test from KafkaService:

package com.snoirot;

import io.quarkus.test.junit.QuarkusTest;
import org.junit.jupiter.api.Test;

@QuarkusTest
class KafkaServiceTest {

  @Test
  void testKafkaService() {

  }
}

So if we run this test, our test will be green and we can see some docker logs:

Test without dev service

First thing, we have to tell Quarkus to not use dev services but only for test profile:

%test.quarkus.kafka.devservices.enabled=false

If I run the test again, there is no docker logs anymore and we have a message saying that the connection to broker could not be etablished.

Setup QuarkusTestResourceLifecycleManager

Now we have to start a Kafka at the beginning of our test.

To do that, we will use QuarkusTestResourceLifecycleManager

We will create a KafkaTestResource that will implement QuarkusTestResourceLifecycleManager and start a Kafka broker there.

We need two dependencies that will contains the Kafka broker:

    <dependency>
      <groupId>com.salesforce.kafka.test</groupId>
      <artifactId>kafka-junit5</artifactId>
      <version>3.2.3</version>
      <scope>test</scope>
    </dependency>
    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>kafka_2.12</artifactId>
      <version>2.8.0</version>
      <scope>test</scope>
    </dependency>

Now we can create KafkaTestResource:

public class KafkaTestResource implements QuarkusTestResourceLifecycleManager {

  public static final Logger LOGGER = Logger.getLogger(KafkaTestResource.class);

  private SharedKafkaTestResource sharedKafkaTestResource;

  @Override
  public Map<String, String> start() {
    sharedKafkaTestResource = new SharedKafkaTestResource().withBrokers(3)
        .withBrokerProperty("auto.create.topics.enable", "true");

    try {
      sharedKafkaTestResource.beforeAll(null);
    } catch (Exception e) {
      LOGGER.error("Error starting kafka broker", e);
    }
    return Map.of( "kafka.bootstrap.servers", sharedKafkaTestResource.getKafkaConnectString());
  }

  @Override
  public void stop() {
    if (sharedKafkaTestResource != null) {
      sharedKafkaTestResource.afterAll(null);
    }
  }
}

This SharedKafkaTestResource would normally be used as an extension in Unit Tests with JUnit 5:

@RegisterExtension
    static final SharedKafkaTestResource sharedKafkaTestResource = new SharedKafkaTestResource()
        .withBrokers(1)
        .withBrokerProperty("auto.create.topics.enable", "true");

and it will call the beforeAll to start the cluster and afterAll to stop it.

As I want to have an Integration Test, I have to call those methods myself.

This KafkaTestResource will return a map that will override the config map:

return Map.of( "kafka.bootstrap.servers", sharedKafkaTestResource.getKafkaConnectString());

It is really useful as the port of the broker is not fixed.

Now i just have to call this resource from my test:

@QuarkusTest
@QuarkusTestResource(value = KafkaTestResource.class, restrictToAnnotatedClass = true)
class KafkaServiceTest {
`

I recommend to use restrictToAnnotatedClass in case you have several Integration Tests.

Let's run the test. I see some exception about some module. I will just add this arguments to my test:

--add-opens java.base/java.lang=ALL-UNNAMED

Let's run the test again and see that the Kafka is started.

Create a test

To create a test, I will use Mockito and Awaitility:

     <dependency>
      <groupId>org.awaitility</groupId>
      <artifactId>awaitility</artifactId>
      <version>4.1.1</version>
      <scope>test</scope>
    </dependency>
    <dependency>
      <groupId>io.quarkus</groupId>
      <artifactId>quarkus-junit5-mockito</artifactId>
      <scope>test</scope>
    </dependency>

NB : We can remove quarkus-junit5 dependency as it is included into quarkus-junit5-mockito

For my test, I will inject a Spy of my KafkaService and see if there is some call of consumeData which will mean my integration is working.

import static org.awaitility.Awaitility.await;

import io.quarkus.test.common.QuarkusTestResource;
import io.quarkus.test.junit.QuarkusTest;
import io.quarkus.test.junit.mockito.InjectSpy;
import java.time.Duration;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;

@QuarkusTest
@QuarkusTestResource(value = KafkaTestResource.class, restrictToAnnotatedClass = true)
class KafkaServiceTest {

  @InjectSpy
  KafkaService kafkaService;

  @Test
  void testKafkaService() {

    var startTime = System.currentTimeMillis();
    await().atMost(Duration.ofSeconds(10)).until(() -> System.currentTimeMillis() - startTime > 9000);

    Mockito.verify(kafkaService, Mockito.atLeastOnce()).consumeData(Mockito.anyString());
  }
}

I use the await to wait 9 seconds as it can take time before the consumer get registered from the cluster.

I run the test, everything is green!

Conclusion

We have seen that we can start a Kafka cluster inside our Integration Tests using kafka-junit5 and a bit of imagination.

I recommend you to go read their github so you can create more complex Integration Tests in the future!