March 25, 2023

Do you take into account unit testing as not sufficient answer for holding the appliance’s reliability and stability? Are you afraid that by some means or someplace there’s a potential bug hiding within the assumption that unit exams ought to cowl all circumstances? And in addition is mocking Kafka not sufficient for challenge necessities? If even one reply is  ‘sure’, then welcome to a pleasant and straightforward information on arrange Integration Exams for Kafka utilizing TestContainers and Embedded Kafka for Spring!

What’s TestContainers?

TestContainers is an open-source Java library specialised in offering all wanted options for the combination and testing of exterior sources. It signifies that we’re capable of mimic an precise database, net server, and even an occasion bus surroundings and deal with that as a dependable place to check app performance. All these fancy options are hooked into docker pictures, outlined as containers. Do we have to take a look at the database layer with precise MongoDB? No worries, we now have a take a look at container for that. We can’t additionally overlook about UI exams – Selenium Container will do something that we really need.
In our case, we are going to concentrate on Kafka Testcontainer.

What’s Embedded Kafka?

Because the title suggests, we’re going to cope with an in-memory Kafka occasion, prepared for use as a standard dealer with full performance. It permits us to work with producers and customers, as normal, making our integration exams light-weight. 

Earlier than we begin

The idea for our take a look at is straightforward – I wish to take a look at Kafka client and producer utilizing two completely different approaches and verify how we will make the most of them in precise circumstances. 

Kafka Messages are serialized utilizing Avro schemas.

Embedded Kafka – Producer Check

The idea is simple – let’s create a easy challenge with the controller, which invokes a service technique to push a Kafka Avro serialized message.


implementation "org.apache.avro:avro:1.10.1"
implementation 'org.springframework.boot:spring-boot-starter-validation'
implementation 'org.springframework.kafka:spring-kafka'

implementation 'org.projectlombok:lombok:1.18.16'

compileOnly 'org.projectlombok:lombok'
annotationProcessor 'org.projectlombok:lombok'
testImplementation 'org.springframework.boot:spring-boot-starter-test'
testImplementation 'org.springframework.kafka:spring-kafka-test'

Additionally value mentioning incredible plugin for Avro. Right here plugins part:

	id 'org.springframework.boot' model '2.6.8'
	id 'io.spring.dependency-management' model '1.0.11.RELEASE'
	id 'java'
	id "com.github.davidmc24.gradle.plugin.avro" model "1.3.0"

Avro Plugin helps schema auto-generating. This can be a must-have.

Hyperlink to plugin:

Now let’s outline the Avro schema:

  "namespace": "com.grapeup.myawesome.myawesomeproducer",
  "sort": "file",
  "title": "RegisterRequest",
  "fields": [
    "name": "id", "type": "long",
    "name": "address", "type": "string", "": "String"


Our ProducerService can be targeted solely on sending messages to Kafka utilizing a template, nothing thrilling about that half. Major performance will be finished simply utilizing this line:

ListenableFuture<SendResult<String, RegisterRequest>> future = this.kafkaTemplate.ship("register-request", kafkaMessage);

We will’t overlook about take a look at properties:

    allow-bean-definition-overriding: true
      group-id: group_id
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.widespread.serialization.StringDeserializer
      value-deserializer: com.grapeup.myawesome.myawesomeconsumer.widespread.CustomKafkaAvroDeserializer
      auto.register.schemas: true
      key-serializer: org.apache.kafka.widespread.serialization.StringSerializer
      value-serializer: com.grapeup.myawesome.myawesomeconsumer.widespread.CustomKafkaAvroSerializer
      particular.avro.reader: true

As we see within the talked about take a look at properties, we declare a customized deserializer/serializer for KafkaMessages. It’s extremely advisable to make use of Kafka with Avro – don’t let JSONs keep object construction, let’s use civilized mapper and object definition like Avro.


public class CustomKafkaAvroSerializer extends KafkaAvroSerializer 
    public CustomKafkaAvroSerializer() 
        tremendous.schemaRegistry = new MockSchemaRegistryClient();

    public CustomKafkaAvroSerializer(SchemaRegistryClient shopper) 
        tremendous(new MockSchemaRegistryClient());

    public CustomKafkaAvroSerializer(SchemaRegistryClient shopper, Map<String, ?> props) 
        tremendous(new MockSchemaRegistryClient(), props);


public class CustomKafkaAvroSerializer extends KafkaAvroSerializer 
    public CustomKafkaAvroSerializer() 
        tremendous.schemaRegistry = new MockSchemaRegistryClient();

    public CustomKafkaAvroSerializer(SchemaRegistryClient shopper) 
        tremendous(new MockSchemaRegistryClient());

    public CustomKafkaAvroSerializer(SchemaRegistryClient shopper, Map<String, ?> props) 
        tremendous(new MockSchemaRegistryClient(), props);

And we now have all the things to start out writing our take a look at.

@ActiveProfiles("take a look at")
@EmbeddedKafka(partitions = 1, matters = "register-request")
class ProducerControllerTest {

All we have to do is add @EmbeddedKafka annotation with listed matters and partitions. Software Context will boot Kafka Dealer with offered configuration identical to that. Understand that @TestInstance needs to be used with particular consideration. Lifecycle.PER_CLASS will keep away from creating the identical objects/context for every take a look at technique. Price checking if exams are too time-consuming.

Shopper<String, RegisterRequest> consumerServiceTest;
void setUp() 
DefaultKafkaConsumerFactory<String, RegisterRequest> client = new DefaultKafkaConsumerFactory<>(kafkaProperties.buildConsumerProperties();

consumerServiceTest = client.createConsumer();

Right here we will declare the take a look at client, primarily based on the Avro schema return sort. All Kafka properties are already offered within the .yml file. That client can be used as a verify if the producer really pushed a message.

Right here is the precise take a look at technique:

void whenValidInput_therReturns200() throws Exception 
        RegisterRequestDto request = RegisterRequestDto.builder()
                .deal with("tempAddress")

        mockMvc.carry out(
                      .content material(objectMapper.writeValueAsBytes(request)))

      ConsumerRecord<String, RegisterRequest> consumedRegisterRequest =  KafkaTestUtils.getSingleRecord(consumerServiceTest, TOPIC_NAME);

        RegisterRequest valueReceived = consumedRegisterRequest.worth();

        assertEquals(12, valueReceived.getId());
        assertEquals("tempAddress", valueReceived.getAddress());

To begin with, we use MockMvc to carry out an motion on our endpoint. That endpoint makes use of ProducerService to push messages to Kafka. KafkaConsumer is used to confirm if the producer labored as anticipated. And that’s it – we now have a completely working take a look at with embedded Kafka.

Check Containers – Shopper Check

TestContainers are nothing else like impartial docker pictures prepared for being dockerized. The next take a look at state of affairs can be enhanced by a MongoDB picture. Why not preserve our knowledge within the database proper after something occurred in Kafka circulate?

Dependencies should not a lot completely different than within the earlier instance. The next steps are wanted for take a look at containers:

testImplementation 'org.testcontainers:junit-jupiter'
	testImplementation 'org.testcontainers:kafka'
	testImplementation 'org.testcontainers:mongodb'

	set('testcontainersVersion', "1.17.1")

		mavenBom "org.testcontainers:testcontainers-bom:$testcontainersVersion"

Let’s focus now on the Shopper half. The take a look at case can be easy – one client service can be chargeable for getting the Kafka message and storing the parsed payload within the MongoDB assortment. All that we have to find out about KafkaListeners, for now, is that annotation:

@KafkaListener(matters = "register-request")

By the performance of the annotation processor, KafkaListenerContainerFactory can be accountable to create a listener on our technique. From this second our technique will react to any upcoming Kafka message with the talked about subject.

Avro serializer and deserializer configs are the identical as within the earlier take a look at.

Relating to TestContainer, we must always begin with the next annotations:

@ActiveProfiles("take a look at")
public class AbstractIntegrationTest {

Throughout startup, all configured TestContainers modules can be activated. It means that we’ll get entry to the complete working surroundings of the chosen supply. As instance:

non-public KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

public static KafkaContainer kafkaContainer = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:6.2.1"));

static MongoDBContainer mongoDBContainer = new MongoDBContainer("mongo:4.4.2").withExposedPorts(27017);

On account of booting the take a look at, we will anticipate two docker containers to start out with the offered configuration.

What is absolutely vital for the mongo container – it offers us full entry to the database utilizing only a easy connection uri. With such a function, we’re ready to have a look what’s the present state in our collections, even throughout debug mode and ready breakpoints.
Have a look additionally on the Ryuk container – it really works like overwatch and checks if our containers have began accurately.

And right here is the final a part of the configuration:

static void dataSourceProperties(DynamicPropertyRegistry registry) 
   registry.add("spring.kafka.bootstrap-servers", kafkaContainer::getBootstrapServers);
   registry.add("spring.kafka.client.bootstrap-servers", kafkaContainer::getBootstrapServers);
   registry.add("spring.kafka.producer.bootstrap-servers", kafkaContainer::getBootstrapServers);
   registry.add("spring.knowledge.mongodb.uri", mongoDBContainer::getReplicaSetUrl);



public void beforeTest() 

           messageListenerContainer -> 
                       .waitForAssignment(messageListenerContainer, 1);


static void tearDown() 

DynamicPropertySource offers us the choice to set all wanted surroundings variables through the take a look at lifecycle. Strongly wanted for any config functions for TestContainers. Additionally, beforeTestClass kafkaListenerEndpointRegistry waits for every listener to get anticipated partitions throughout container startup.

And the final a part of the Kafka take a look at containers journey – the principle physique of the take a look at:

public void containerStartsAndPublicPortIsAvailable() throws Exception 
   writeToTopic("register-request", RegisterRequest.newBuilder().setId(123).setAddress("dummyAddress").construct());

   //Look ahead to KafkaListener
   Assertions.assertEquals(1, taxiRepository.findAll().dimension());

non-public KafkaProducer<String, RegisterRequest> createProducer() 
   return new KafkaProducer<>(kafkaProperties.buildProducerProperties());

non-public void writeToTopic(String topicName, RegisterRequest... registerRequests) 

   strive (KafkaProducer<String, RegisterRequest> producer = createProducer())
               .forEach(registerRequest -> 
                           ProducerRecord<String, RegisterRequest> file = new ProducerRecord<>(topicName, registerRequest);

The customized producer is chargeable for writing our message to KafkaBroker. Additionally, it is suggested to offer a while for customers to deal with messages correctly. As we see, the message was not simply consumed by the listener, but in addition saved within the MongoDB assortment.


As we will see, present options for integration exams are fairly simple to implement and keep in initiatives. There is no such thing as a level in holding simply unit exams and relying on all strains lined as an indication of code/logic high quality. Now the query is, ought to we use an Embedded answer or TestContainers? I counsel initially specializing in the phrase “Embedded”. As an ideal integration take a look at, we wish to get an virtually splendid copy of the manufacturing surroundings with all properties/options included. In-memory options are good, however largely, not sufficient for giant enterprise initiatives. Undoubtedly, the benefit of Embedded providers is the straightforward approach to implement such exams and keep configuration, simply when something occurs in reminiscence.
TestContainers on the first sight may appear like overkill, however they offer us crucial function, which is a separate surroundings. We don’t should even depend on current docker pictures – if we wish we will use customized ones. This can be a big enchancment for potential take a look at situations.
What about Jenkins? There is no such thing as a purpose to be afraid additionally to make use of TestContainers in Jenkins. I firmly advocate checking TestContainers documentation on how simply we will arrange the configuration for Jenkins brokers.
To sum up – if there is no such thing as a blocker or any undesirable situation for utilizing TestContainers, then don’t hesitate. It’s all the time good to maintain all providers managed and secured with integration take a look at contracts.