Всем привет! Меня зовут Виталий, я разработчик в компании Web3Tech. В этом посте я представлю основные концепции и конструкции платформы Spring Cloud Stream для поддержки и работы с брокерами сообщений Kafka, с полным циклом их контекстного unit-тестирования. Мы используем такую схему в своем проекте всероссийского электронного голосования на блокчейн-платформе Waves Enterprise.
Являясь частью группы проектов Spring Cloud, Spring Cloud Stream основан на Spring Boot и использует Spring Integration для обеспечения связи с брокерами сообщений. При этом он легко интегрируется с различными брокерами сообщений и требует минимальной конфигурации для создания event-driven или message-driven микросервисов.
Конфигурация и зависимости
Для начала нам нужно добавить зависимость spring-cloud-starter-stream-kafka в build.gradle:
dependencies { implementation(kotlin("stdlib")) implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:$kotlinCoroutinesVersion") implementation("com.fasterxml.jackson.module:jackson-module-kotlin") implementation("org.springframework.boot:spring-boot-starter-web") implementation("org.springframework.cloud:spring-cloud-starter-stream-kafka") testImplementation("org.springframework.boot:spring-boot-starter-test") testImplementation("org.springframework.cloud:spring-cloud-stream-test-support") testImplementation("org.springframework.kafka:spring-kafka-test:springKafkaTestVersion")}
В конфигурацию проекта Spring Cloud Stream необходимо включить URL Kafka-брокера, имя очереди (топик) и другие параметры биндинга. Вот пример YAML-конфигурации для сервиса application.yaml:
spring: application: name: cloud-stream-binding-kafka-app cloud: stream: kafka: binder: brokers: 0.0.0.0:8080 configuration: auto-offset-reset: latest bindings: customChannel: #Channel name destination: 0.0.0.0:8080 #Destination to which the message is sent (topic) group: input-group-N contentType: application/json consumer: max-attempts: 1 autoCommitOffset: true autoCommitOnError: false
Концепция и классы
По сути, мы имеем дело с сервисом, построенным на Spring Cloud Stream, который прослушивает входящую очередь, используя биндинги (SpringCloudStreamBindingKafkaApp.kt):
@EnableBinding(ProducerBinding::class)@SpringBootApplication class SpringCloudStreamBindingKafkaApp fun main(args: Array<String>) { SpringApplication.run(SpringCloudStreamBindingKafkaApp::class.java, *args) }
Аннотация @EnableBinding указывает сервису на биндинг как входящего, так и исходящего канала.
Здесь необходимо уточнить ряд концепций.
Binding интерфейс, в котором описаны входящие и
исходящие каналы.
Binder имплементация middleware для сообщений.
Channel представляет канал для передачи сообщений
между middleware и приложением.
StreamListeners методы обработки сообщений в виде
бинов (beans), которые будут автоматически вызваны после того, как
MessageConverter осуществит сериализацию или десериализацию между
событиями в middleware и типами объектов в домене DTO.
Message Schema схемы, используемые для
сериализации и десериализации сообщений. Могут быть прочитаны из
источника или динамически загружены.
Тестирование
Чтобы протестировать сообщение и операции send/receive, нам нужно создать как минимум одного producer и одного consumer. Вот простейший пример того, как это можно сделать в Spring Cloud Stream.
Инстанс бина Producer будет отправлять сообщение в топик Kafka, используя биндер (ProducerBinding.kt):
interface ProducerBinding { @Output(BINDING_TARGET_NAME) fun messageChannel(): MessageChannel}
Инстанс бина Сonsumer будет слушать топик Kafka и получать сообщения.
ConsumerBinding.kt:
interface ConsumerBinding { companion object { const val BINDING_TARGET_NAME = "customChannel" } @Input(BINDING_TARGET_NAME) fun messageChannel(): MessageChannel}
Consumer.kt:
@EnableBinding(ConsumerBinding::class)class Consumer(val messageService: MessageService) { @StreamListener(target = ConsumerBinding.BINDING_TARGET_NAME) fun process( @Payload message: Map<String, Any?>, @Header(value = KafkaHeaders.OFFSET, required = false) offset: Int? ) { messageService.consume(message) }}
Мы создали брокер Kafka с топиком. Для тестирования будем использовать встроенную Kafka, доступную нам с зависимостью spring-kafka-test.
Функциональное тестирование с MessageCollector
Мы имеем дело с имплементацией биндера, позволяющей взаимодействовать с каналами и получать сообщения. Отправим сообщение в канал ProducerBinding и затем получим его в виде payloadProducerTest.kt:
@SpringBootTestclass ProducerTest { @Autowired lateinit var producerBinding: ProducerBinding @Autowired lateinit var messageCollector: MessageCollector @Test fun `should produce somePayload to channel`() { // ARRANGE val request = mapOf(1 to "foo", 2 to "bar", "three" to 10101) // ACTproducerBinding.messageChannel().send(MessageBuilder.withPayload(request).build()) val payload = messageCollector.forChannel(producerBinding.messageChannel()) .poll() .payload // ASSERT val payloadAsMap = jacksonObjectMapper().readValue(payload.toString(), Map::class.java) assertTrue(request.entries.stream().allMatch { re -> re.value == payloadAsMap[re.key.toString()] }) messageCollector.forChannel(producerBinding.messageChannel()).clear() }}
Тестирование с брокером Embedded Kafka
Используем аннотацию @ClassRule для создания брокера. Так мы сможем поднять сервера Kafka и Zookeeper на случайном порте перед началом теста и выключить их, когда тест завершится. Это избавляет нас от необходимости в рабочем инстансе Kafka и Zookeper на всё время проведения теста (ConsumerTest.kt):
@SpringBootTest@ActiveProfiles("test")@EnableAutoConfiguration(exclude = [TestSupportBinderAutoConfiguration::class])@EnableBinding(ProducerBinding::class)class ConsumerTest { @Autowired lateinit var producerBinding: ProducerBinding @Autowired lateinit var objectMapper: ObjectMapper @MockBean lateinit var messageService: MessageService companion object { @ClassRule @JvmField var embeddedKafka = EmbeddedKafkaRule(1, true, "any-name-of-topic") } @Test fun `should consume via txConsumer process`() { // ACT val request = mapOf(1 to "foo", 2 to "bar") producerBinding.messageChannel().send(MessageBuilder.withPayload(request) .setHeader("someHeaderName", "someHeaderValue") .build()) // ASSERT val requestAsMap = objectMapper.readValue<Map<String, Any?>>(objectMapper.writeValueAsString(request)) runBlocking { delay(20) verify(messageService).consume(requestAsMap) } }}
Заключение
В этом посте я продемонстрировал возможности Spring Cloud Stream и использования его с Kafka. Spring Cloud Stream предлагает удобный интерфейс с упрощенными нюансами настройки брокера, быстро внедряется, стабильно работает и поддерживает современные популярные брокеры, такие как Kafka. По итогам я привел ряд примеров с unit-тестированием на основе EmbeddedKafkaRule с использованием MessageCollector.
Все исходники можно найти на Github. Спасибо за прочтение!
