How to Set Up Kafka Integration Test – Grape Up

Do you consider device screening as not more than enough answer for maintaining the application’s trustworthiness and steadiness? Are you fearful that in some way or someplace there is a probable bug hiding in the assumption that device assessments should really deal with all conditions? And also is mocking Kafka not sufficient for project prerequisites? If even 1 answer is  ‘yes’, then welcome to a good and uncomplicated information on how to established up Integration Checks for Kafka applying TestContainers and Embedded Kafka for Spring!

What is TestContainers?

TestContainers is an open-resource Java library specialised in supplying all desired answers for the integration and testing of external sources. It usually means that we are equipped to mimic an real database, world wide web server, or even an function bus surroundings and deal with that as a dependable spot to check app functionality. All these fancy characteristics are hooked into docker pictures, described as containers. Do we have to have to exam the databases layer with precise MongoDB? No worries, we have a test container for that. We can not also neglect about UI checks – Selenium Container will do anything at all that we essentially need to have.
In our case, we will concentration on Kafka Testcontainer.

What is Embedded Kafka?

As the name suggests, we are likely to deal with an in-memory Kafka occasion, prepared to be employed as a ordinary broker with total performance. It permits us to get the job done with producers and customers, as common, earning our integration checks light-weight. 

Right before we start

The notion for our test is simple – I would like to exam Kafka buyer and producer making use of two diverse strategies and look at how we can use them in genuine scenarios. 

Kafka Messages are serialized employing Avro schemas.

Embedded Kafka – Producer Check

The principle is uncomplicated – let us produce a straightforward venture with the controller, which invokes a service method to drive a Kafka Avro serialized message.

Dependencies:

dependencies 
implementation "org.apache.avro:avro:1.10.1"
implementation("io.confluent:kafka-avro-serializer:6.1.")
implementation 'org.springframework.boot:spring-boot-starter-validation'
implementation 'org.springframework.kafka:spring-kafka'
implementation('org.springframework.cloud:spring-cloud-stream:3.1.1')
implementation('org.springframework.cloud:spring-cloud-stream-binder-kafka:3.1.1')

implementation('org.springframework.boot:spring-boot-starter-net:2.4.3')
implementation 'org.projectlombok:lombok:1.18.16'

compileOnly 'org.projectlombok:lombok'
annotationProcessor 'org.projectlombok:lombok'
testImplementation('org.springframework.cloud:spring-cloud-stream-examination-support:3.1.1')
testImplementation 'org.springframework.boot:spring-boot-starter-test'
testImplementation 'org.springframework.kafka:spring-kafka-test'

Also value mentioning fantastic plugin for Avro. In this article plugins portion:

plugins 
id 'org.springframework.boot' model '2.6.8'
id 'io.spring.dependency-management' variation '1..11.RELEASE'
id 'java'
id "com.github.davidmc24.gradle.plugin.avro" model "1.3."

Avro Plugin supports schema car-creating. This is a need to-have.

Url to plugin: https://github.com/davidmc24/gradle-avro-plugin

Now let us outline the Avro schema:

  "namespace": "com.grapeup.myawesome.myawesomeproducer",
  "style": "report",
  "name": "RegisterRequest",
  "fields": [
    "name": "id", "type": "long",
    "name": "address", "type": "string", "avro.java.string": "String"
    

  ]

Our ProducerService will be centered only on sending messages to Kafka employing a template, nothing fascinating about that component. Key performance can be performed just employing this line:

ListenableFuture> long term = this.kafkaTemplate.deliver("register-ask for", kafkaMessage)

We can’t forget about about check qualities:

spring:
  principal:
    allow-bean-definition-overriding: correct
  kafka:
    client:
      team-id: group_id
      car-offset-reset: earliest
      crucial-deserializer: org.apache.kafka.frequent.serialization.StringDeserializer
      price-deserializer: com.grapeup.myawesome.myawesomeconsumer.common.CustomKafkaAvroDeserializer
    producer:
      auto.register.schemas: real
      vital-serializer: org.apache.kafka.frequent.serialization.StringSerializer
      worth-serializer: com.grapeup.myawesome.myawesomeconsumer.common.CustomKafkaAvroSerializer
    houses:
      distinct.avro.reader: real

As we see in the pointed out take a look at houses, we declare a personalized deserializer/serializer for KafkaMessages. It is really suggested to use Kafka with Avro – really don’t let JSONs retain object composition, let’s use civilized mapper and object definition like Avro.

Serializer:

community course CustomKafkaAvroSerializer extends KafkaAvroSerializer 
    community CustomKafkaAvroSerializer() 
        tremendous()
        super.schemaRegistry = new MockSchemaRegistryClient()
    

    community CustomKafkaAvroSerializer(SchemaRegistryClient consumer) 
        tremendous(new MockSchemaRegistryClient())
    

    community CustomKafkaAvroSerializer(SchemaRegistryClient consumer, Map props) 
        super(new MockSchemaRegistryClient(), props)
    

Deserializer:

community course CustomKafkaAvroSerializer extends KafkaAvroSerializer 
    community CustomKafkaAvroSerializer() 
        super()
        tremendous.schemaRegistry = new MockSchemaRegistryClient()
    

    public CustomKafkaAvroSerializer(SchemaRegistryClient client) 
        super(new MockSchemaRegistryClient())
    

    general public CustomKafkaAvroSerializer(SchemaRegistryClient shopper, Map props) 
        tremendous(new MockSchemaRegistryClient(), props)
    

And we have everything to start off crafting our check.

@ExtendWith(SpringExtension.course)
@SpringBootTest
@AutoConfigureMockMvc
@TestInstance(TestInstance.Lifecycle.For each_Class)
@ActiveProfiles("test")
@EmbeddedKafka(partitions = 1, matters = "sign up-request")
course ProducerControllerTest {

All we have to have to do is include @EmbeddedKafka annotation with stated matters and partitions. Application Context will boot Kafka Broker with delivered configuration just like that. Retain in intellect that @TestInstance really should be used with specific thing to consider. Lifecycle.For each_Class will stay away from building the exact same objects/context for just about every take a look at approach. Value examining if tests are much too time-consuming.

Buyer consumerServiceTest

@BeforeEach
void Set up() 
DefaultKafkaConsumerFactory shopper = new DefaultKafkaConsumerFactory<>(kafkaProperties.buildConsumerProperties()

consumerServiceTest = customer.createConsumer()
consumerServiceTest.subscribe(Collections.singletonList(Topic_Name))

Below we can declare the exam consumer, dependent on the Avro schema return type. All Kafka houses are already delivered in the .yml file. That shopper will be made use of as a test if the producer truly pushed a information.

Here is the true take a look at method:

@Take a look at
void whenValidInput_therReturns200() throws Exception 
        RegisterRequestDto ask for = RegisterRequestDto.builder()
                .id(12)
                .deal with("tempAddress")
                .develop()

        mockMvc.accomplish(
                submit("/sign up-request")
                      .contentType("application/json")
                      .material(objectMapper.writeValueAsBytes(request)))
                .andExpect(standing().isOk())

      ConsumerRecord consumedRegisterRequest =  KafkaTestUtils.getSingleRecord(consumerServiceTest, Topic_Title)

        RegisterRequest valueReceived = consumedRegisterRequest.worth()

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

Initially of all, we use MockMvc to conduct an motion on our endpoint. That endpoint utilizes ProducerService to press messages to Kafka. KafkaConsumer is applied to verify if the producer worked as envisioned. And that is it – we have a totally operating exam with embedded Kafka.

Check Containers – Customer Take a look at

TestContainers are nothing at all else like impartial docker photos ready for being dockerized. The next check state of affairs will be increased by a MongoDB picture. Why not keep our info in the databases correct right after just about anything occurred in Kafka stream?

Dependencies are not substantially distinctive than in the prior illustration. The next measures are needed for take a look at containers:

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

ext 
set('testcontainersVersion', "1.17.1")


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

Let’s emphasis now on the Client element. The examination case will be easy – one purchaser support will be accountable for having the Kafka concept and storing the parsed payload in the MongoDB assortment. All that we need to have to know about KafkaListeners, for now, is that annotation:

@KafkaListener(topics = "sign up-ask for")

By the operation of the annotation processor, KafkaListenerContainerFactory will be liable to build a listener on our technique. From this instant our approach will react to any forthcoming Kafka information with the stated subject.

Avro serializer and deserializer configs are the same as in the preceding check.

Regarding TestContainer, we ought to commence with the pursuing annotations:

@SpringBootTest
@ActiveProfiles("test")
@Testcontainers
community class AbstractIntegrationTest {

During startup, all configured TestContainers modules will be activated. It usually means that we will get obtain to the total functioning environment of the picked source. As illustration:

@Autowired
non-public KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry

@Container
community static KafkaContainer kafkaContainer = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:6.2.1"))

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

As a end result of booting the exam, we can be expecting two docker containers to start out with the provided configuration.

What is really important for the mongo container – it provides us complete access to the database using just a straightforward relationship uri. With such a function, we are ready to take a search what is the existing state in our collections, even for the duration of debug manner and prepared breakpoints.
Get a look also at the Ryuk container – it is effective like overwatch and checks if our containers have commenced appropriately.

And below is the previous aspect of the configuration:

@DynamicPropertySource
static void dataSourceProperties(DynamicPropertyRegistry registry) 
   registry.add("spring.kafka.bootstrap-servers", kafkaContainer::getBootstrapServers)
   registry.include("spring.kafka.purchaser.bootstrap-servers", kafkaContainer::getBootstrapServers)
   registry.increase("spring.kafka.producer.bootstrap-servers", kafkaContainer::getBootstrapServers)
   registry.incorporate("spring.data.mongodb.uri", mongoDBContainer::getReplicaSetUrl)


static 
   kafkaContainer.commence()
   mongoDBContainer.begin()

   mongoDBContainer.waitingFor(Wait around.forListeningPort()
           .withStartupTimeout(Period.ofSeconds(180L)))


@BeforeTestClass
general public void beforeTest() 

   kafkaListenerEndpointRegistry.getListenerContainers().forEach(
           messageListenerContainer -> 
               ContainerTestUtils
                       .waitForAssignment(messageListenerContainer, 1)

           
   )


@AfterAll
static void tearDown() 
   kafkaContainer.prevent()
   mongoDBContainer.end()

DynamicPropertySource offers us the selection to set all required natural environment variables for the duration of the test lifecycle. Strongly desired for any config purposes for TestContainers. Also, beforeTestClass kafkaListenerEndpointRegistry waits for every listener to get expected partitions during container startup.

And the past portion of the Kafka examination containers journey – the key body of the check:

@Exam
general public void containerStartsAndPublicPortIsAvailable() throws Exception 
   writeToTopic("sign-up-request", RegisterRequest.newBuilder().setId(123).setAddress("dummyAddress").build())

   //Wait around for KafkaListener
   TimeUnit.SECONDS.rest(5)
   Assertions.assertEquals(1, taxiRepository.findAll().dimension())



private KafkaProducer createProducer() 
   return new KafkaProducer<>(kafkaProperties.buildProducerProperties())


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

   attempt (KafkaProducer producer = createProducer()) 
       Arrays.stream(registerRequests)
               .forEach(registerRequest -> 
                           ProducerRecord record = new ProducerRecord<>(topicName, registerRequest)
                           producer.send(document)
                       
               )
   

The customized producer is dependable for producing our information to KafkaBroker. Also, it is advised to give some time for shoppers to manage messages adequately. As we see, the message was not just consumed by the listener, but also stored in the MongoDB assortment.

Conclusions

As we can see, existing remedies for integration exams are really easy to put into action and manage in tasks. There is no issue in retaining just device checks and counting on all traces coated as a signal of code/logic quality. Now the problem is, must we use an Embedded option or TestContainers? I recommend 1st of all focusing on the phrase “Embedded”. As a best integration check, we want to get an almost best copy of the production surroundings with all properties/attributes included. In-memory answers are good, but mainly, not adequate for substantial organization jobs. Undoubtedly, the benefit of Embedded companies is the quick way to apply these types of checks and retain configuration, just when nearly anything transpires in memory.
TestContainers at the first sight may possibly search like overkill, but they give us the most vital attribute, which is a separate natural environment. We don’t have to even rely on existing docker illustrations or photos – if we want we can use custom made types. This is a large enhancement for potential take a look at situations.
What about Jenkins? There is no explanation to be fearful also to use TestContainers in Jenkins. I firmly endorse examining TestContainers documentation on how quickly we can set up the configuration for Jenkins agents.
To sum up – if there is no blocker or any undesired problem for employing TestContainers, then really do not wait. It is always great to keep all products and services managed and secured with integration test contracts.