The expertise of working Kafka in take a look at situations has reached a excessive stage of comfort because of using Take a look at containers and enhanced help in Spring Boot 3.1 with the @ServiceConnection
annotation. Nonetheless, writing and sustaining integration checks with Kafka stays a problem. This text describes an strategy that considerably simplifies the testing course of by making certain take a look at isolation and offering a set of instruments to realize this purpose. With the profitable implementation of isolation, Kafka checks will be organized in such a method that on the stage of consequence verification, there may be full entry to all messages which have arisen throughout the take a look at, thereby avoiding the necessity for pressured ready strategies equivalent to Thread.sleep()
.
This technique is appropriate to be used with Take a look at containers, Embedded Kafka, or different strategies of working the Kafka service (e.g., a neighborhood occasion).
Take a look at Isolation
As detailed within the article Eradicating Non-Determinism in Assessments, clear management over the testing setting is essential for the dependable execution of checks. This ensures that every take a look at begins from a identified state. An instance may very well be a scenario the place one take a look at creates knowledge in a database and fails to scrub it up afterward, negatively impacting the execution of subsequent checks that anticipate a special database state.
Varied strategies will be utilized to realize take a look at isolation, together with:
- Restoring the system’s unique state earlier than every take a look at run.
- Obligatory knowledge cleanup after the execution of every take a look at to stop affect on subsequent checks.
Isolation in Kafka Testing
Restoring Kafka’s preliminary setting state from scratch for every take a look at state of affairs will be achieved by restarting Kafka. This selection is easy by way of implementation however expensive by way of startup time. There are strategies to hurry up this course of (for extra particulars, you may examine working Kafka on GraalVM), however this text suggests contemplating an choice the place we work with a typical Kafka occasion throughout all take a look at situations.
This strategy poses sure challenges: if one take a look at sends messages to a subject and neglects the actual fact of their receipt and processing, it may hurt the execution of one other take a look at that will depend on a special matter state.
To make sure isolation, it’s mandatory that within the case of sending a message to a subject being listened to by the identical software, all processes initiated by these messages are accomplished earlier than the tip of the take a look at state of affairs.
Implementation
Take into account the instance of a hypothetical Telegram bot that redirects requests to the OpenAI API and sends responses to customers.
For simplicity, the interplay contracts with the providers are described in a simplified type to spotlight the primary logic of the operation. Beneath is a sequence diagram displaying the structure of the applying. I perceive that the design might increase questions from a system structure perspective, however please bear with me—the primary purpose right here is to show the strategy to isolation in testing.
Beneath is a diagram illustrating the testing strategy:
The important thing function of the proposed answer is the strict division of take a look at code into phases similar to the Organize-Act-Assert sample. You may learn extra about this strategy within the article Ordering Chaos: Arranging HTTP Request Testing in Spring.
To realize isolation, it’s critically vital to take care of the next relationships between the important thing parts of the scheme (numbers correspond to these indicated within the diagram):
- State of affairs setup (Organize) happens earlier than State of affairs execution (Act).
- The message is distributed and (3) confirmed by the partition chief earlier than the ServiceA’s request processing is taken into account full.
- Handbook offset administration with commit happens solely in any case processing by ServiceB or ServiceA is absolutely accomplished.
- State of affairs execution (Act) happens earlier than Outcome verification (Assert).
This strategy ensures that by the point of consequence verification, all processes throughout the take a look at state of affairs are accomplished, and all messages are despatched and acquired, making certain the take a look at setting is in a identified and closing state.
State of affairs Setup (Organize) Happens Earlier than State of affairs Execution (Act)
The purpose of this stage is to organize all the pieces mandatory for the take a look at state of affairs. Within the context of our instance, the primary elements of the take a look at setting embody the applying context, HTTP request mocks, Kafka, and File Captor.
Concerning integration with Kafka, it’s vital to make sure that all shoppers are able to obtain messages. This verification is applied within the KafkaSupport#waitForPartitionAssignment technique. The answer relies on the unique ContainerTestUtils
from the org.springframework.kafka:spring-kafka-test
library, modified in keeping with the described utilization state of affairs. This technique ensures that not less than one partition shall be assigned to every Kafka client. This suggests a limitation: there have to be just one partition per matter within the take a look at setting, though this limitation is a consequence of the present implementation of the tactic and will be modified.
Utilizing a typical Kafka occasion requires setting the auto.offset.reset = newest
parameter for shoppers. For Spring functions, that is executed as follows:
spring.kafka.client.auto-offset-reset=newest
File Captor is a key component of this answer. Its job is to “catch” messages from the desired subjects within the configuration and supply entry to them for the consequence verification step of the take a look at state of affairs. Technically, it’s a easy client for a Kafka matter with a message storage mechanism and an entry interface. The File Captor code is obtainable within the venture repository.
The present implementation of File Captor gives using the message key for figuring out messages associated to a particular take a look at case. That is helpful in methods the place distinctive identifiers are current, equivalent to a shopper ID or a course of identifier within the area mannequin. Utilizing such identifiers as the important thing permits for successfully grouping and monitoring all messages associated to the identical take a look at state of affairs, even when they’re distributed throughout completely different subjects or processed by completely different system elements.
Synchronous Message Sending With Acknowledgment
The purpose is to implement synchronous message sending to Kafka with acknowledgment from the partition chief. To realize this, the acks = 1
parameter have to be set for the producer. Within the context of a Spring software, this setting is specified as follows:
spring.kafka.producer.acks=1
When utilizing KafkaTemplate
for sending messages, you will need to make sure the synchronicity of sending, as this element gives solely an asynchronous interface org.springframework.kafka.core.KafkaTemplate#ship(org.springframework.messaging.Message>) return CompletableFuture
. The next strategy can be utilized for synchronous sending:
kafkaTemplate.ship(message).get()
This ensures that the message sending shall be accomplished synchronously, ready for Kafka’s acknowledgment earlier than continuing with this system execution.
Handbook Offset Administration
Handbook offset administration with commit signifies that the message client will commit the message processing solely after their full processing. On this context, the offset for topicA
will solely be dedicated after the message has been efficiently despatched to topicB
and the corresponding acknowledgment has been acquired.
To implement this logic, it’s essential to disable the automated offset commit for shoppers by setting the allow.auto.commit = false
parameter. Within the context of a Spring software, that is configured as follows:
spring.kafka.client.enable-auto-commit=false
Moreover, the commit mechanism needs to be configured in order that the offset is dedicated after processing every particular person message, which is achieved by setting the parameter:
spring.kafka.listener.ack-mode=document
State of affairs Execution (Act) Happens Earlier than Outcome Verification (Assert)
Earlier than beginning the consequence verification stage, it’s mandatory to make sure that all processes associated to the state of affairs are accomplished, and all messages are despatched and acquired, making certain the take a look at setting transitions to a “known” closing state. Due to the previous levels, now we have ensured adherence to the happens-before precept between the actions of producers and shoppers, in addition to between all processing throughout the software. At this stage, it’s required to carry out a verify for offset commits for every partition for every group of shoppers.
To hold out this verification, the answer introduced within the technique pw.avvero.emk.KafkaSupport#waitForPartitionOffsetCommit can be utilized.
Outcome Verification (Assert)
The ultimate stage includes analyzing the outcomes of the take a look at state of affairs. This course of consists of checking the state of mocks and analyzing the messages caught in RecordCaptor.
Key Answer Components Summarized
Here’s a transient abstract of the important thing elements of the proposed answer for efficient testing with Kafka:
- One partition per matter.
- The coverage for message studying begin for shoppers:
spring.kafka.client.auto-offset-reset=newest
. - Acknowledgment coverage for producers:
spring.kafka.producer.acks=1
. - Synchronous message sending:
kafkaTemplate.ship(message).get()
. - Handbook offset management:
spring.kafka.client.enable-auto-commit=false, spring.kafka.listener.ack-mode=document
. - Ready for partition task earlier than beginning the take a look at state of affairs:
pw.avvero.emk.KafkaSupport#waitForPartitionAssignment
. - Ready for offset dedication earlier than verifying take a look at outcomes:
pw.avvero.emk.KafkaSupport#waitForPartitionOffsetCommit
.
Outcome
The applying code is obtainable within the examples module. The take a look at code is as follows.
def "User Message Processing with OpenAI"() {
setup:
(1) KafkaSupport.waitForPartitionAssignment(applicationContext)
and:
def openaiRequestCaptor = new RequestCaptor()
(2) restMock.anticipate(manyTimes(), requestTo("https://api.openai.com/v1/chat/completions"))
.andExpect(technique(HttpMethod.POST))
.andExpect(openaiRequestCaptor)
.andRespond(withSuccess('{"content": "Hi, how can i help you?"}', MediaType.APPLICATION_JSON))
and:
def telegramRequestCaptor = new RequestCaptor()
(3) restMock.anticipate(manyTimes(), requestTo("https://api.telegram.org/sendMessage"))
.andExpect(technique(HttpMethod.POST))
.andExpect(telegramRequestCaptor)
.andRespond(withSuccess('{}', MediaType.APPLICATION_JSON))
when:
(4) mockMvc.carry out(publish("/telegram/webhook")
.contentType(APPLICATION_JSON_VALUE)
.content material("""{
"message": {
"from": {
"id": 10000000
},
"chat": {
"id": 20000000
},
"text": "Hello!"
}
}""".toString())
.settle for(APPLICATION_JSON_VALUE))
.andExpect(standing().isOk())
(5) KafkaSupport.waitForPartitionOffsetCommit(applicationContext)
then:
(6) openaiRequestCaptor.occasions == 1
JSONAssert.assertEquals("""{
"content": "Hello!"
}""", openaiRequestCaptor.bodyString, false)
and:
telegramRequestCaptor.occasions == 1
JSONAssert.assertEquals("""{
"chatId": "20000000",
"text": "Hi, how can i help you?"
}""", telegramRequestCaptor.bodyString, false)
}
Listed below are the important thing steps described within the take a look at state of affairs:
- Ready for partition task earlier than beginning the take a look at state of affairs.
- Mocking requests to OpenAI.
- Mocking requests to Telegram.
- Executing the take a look at state of affairs.
- Ready for offset affirmation.
- Outcome verification.
Extra checks, together with situations with intensive message sending and using the @RetryableTopic
mechanism for retries, are additionally accessible within the venture repository, offering alternatives for examine and adaptation to your personal improvement wants.
Conclusion
Profitable testing of interactions with Kafka requires a cautious strategy to check isolation and setting management. The usage of Take a look at containers and the capabilities of Spring Boot 3.1 vastly simplifies this course of, and the applying of the proposed methods and instruments permits builders to give attention to the applying logic, making improvement extra environment friendly and fewer liable to errors.
Thanks for listening to the article, and good luck in your pursuit of writing handy and dependable checks!