From 0efa3076302d44234125ce2a021cbb440edcda99 Mon Sep 17 00:00:00 2001 From: Christian Tzolov Date: Thu, 3 Aug 2023 15:45:01 +0200 Subject: [PATCH] Add smoke test for Spring Kafka with AVRO Closes gh-11 --- integration/spring-kafka-avro/README.adoc | 1 + integration/spring-kafka-avro/build.gradle | 30 +++ .../spring-kafka-avro/docker-compose.yml | 25 +++ .../kafka/KafkaAvroApplicationCrTests.java | 46 +++++ .../src/main/avro/Thing1.avsc | 10 + .../src/main/avro/Thing2.avsc | 10 + .../src/main/avro/Thing3.avsc | 10 + .../src/main/avro/Thing4.avsc | 10 + .../src/main/avro/Thing5.avsc | 10 + .../src/main/avro/Thing6.avsc | 10 + .../src/main/avro/Thing7.avsc | 10 + .../src/main/avro/Thing8.avsc | 10 + .../example/kafka/KafkaAvroApplication.java | 193 ++++++++++++++++++ .../com/example/kafka/ThingAvroSerde.java | 117 +++++++++++ .../src/main/resources/application.properties | 4 + 15 files changed, 496 insertions(+) create mode 100644 integration/spring-kafka-avro/README.adoc create mode 100644 integration/spring-kafka-avro/build.gradle create mode 100644 integration/spring-kafka-avro/docker-compose.yml create mode 100644 integration/spring-kafka-avro/src/appTest/java/com/example/kafka/KafkaAvroApplicationCrTests.java create mode 100644 integration/spring-kafka-avro/src/main/avro/Thing1.avsc create mode 100644 integration/spring-kafka-avro/src/main/avro/Thing2.avsc create mode 100644 integration/spring-kafka-avro/src/main/avro/Thing3.avsc create mode 100644 integration/spring-kafka-avro/src/main/avro/Thing4.avsc create mode 100644 integration/spring-kafka-avro/src/main/avro/Thing5.avsc create mode 100644 integration/spring-kafka-avro/src/main/avro/Thing6.avsc create mode 100644 integration/spring-kafka-avro/src/main/avro/Thing7.avsc create mode 100644 integration/spring-kafka-avro/src/main/avro/Thing8.avsc create mode 100644 integration/spring-kafka-avro/src/main/java/com/example/kafka/KafkaAvroApplication.java create mode 100644 integration/spring-kafka-avro/src/main/java/com/example/kafka/ThingAvroSerde.java create mode 100644 integration/spring-kafka-avro/src/main/resources/application.properties diff --git a/integration/spring-kafka-avro/README.adoc b/integration/spring-kafka-avro/README.adoc new file mode 100644 index 0000000..c389ec2 --- /dev/null +++ b/integration/spring-kafka-avro/README.adoc @@ -0,0 +1 @@ +Verifies that Spring Kafka stream with AVRO encoding is CR compatible. diff --git a/integration/spring-kafka-avro/build.gradle b/integration/spring-kafka-avro/build.gradle new file mode 100644 index 0000000..e3bb5ea --- /dev/null +++ b/integration/spring-kafka-avro/build.gradle @@ -0,0 +1,30 @@ +plugins { + id "java" + id "org.springframework.boot" + id "org.springframework.cr.smoke-test" + id "com.github.davidmc24.gradle.plugin.avro" version "1.3.0" +} + +dependencies { + implementation(platform(org.springframework.boot.gradle.plugin.SpringBootPlugin.BOM_COORDINATES)) + implementation("org.springframework.boot:spring-boot-starter") + + // TODO: update to non snapshot spring-kafka version when available. + // The spring-kafka 3.0.10+ is required due to https://github.com/spring-projects/spring-kafka/issues/2760 + implementation("org.springframework.kafka:spring-kafka:3.0.10-SNAPSHOT") + + implementation("com.fasterxml.jackson.core:jackson-databind") + implementation("org.apache.avro:avro:1.11.0") + + implementation("org.crac:crac:$cracVersion") + implementation(project(":cr-listener")) + + testImplementation("org.springframework.boot:spring-boot-starter-test") + + appTestImplementation(project(":cr-smoke-test-support")) + appTestImplementation("org.awaitility:awaitility:4.2.0") +} + +tasks.named("checkFormatMain").configure { + exclude { it.file.absolutePath.contains("generated-main-avro-java") } +} diff --git a/integration/spring-kafka-avro/docker-compose.yml b/integration/spring-kafka-avro/docker-compose.yml new file mode 100644 index 0000000..5850c55 --- /dev/null +++ b/integration/spring-kafka-avro/docker-compose.yml @@ -0,0 +1,25 @@ +version: '3' +services: + kafka: + image: wurstmeister/kafka + hostname: kafka + ports: + - '9092' + volumes: + - /var/run/docker.sock:/var/run/docker.sock + environment: + KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181 + KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT + KAFKA_LISTENERS: PLAINTEXT://:29092,PLAINTEXT_HOST://:9092 + KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:_{PORT_COMMAND} + KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT + PORT_COMMAND: "docker ps | egrep 'kafka(-|_)1' | cut -d: -f 2 | cut -d- -f 1" + depends_on: + - zookeeper + + zookeeper: + image: wurstmeister/zookeeper + ports: + - '2181' + environment: + - KAFKA_ADVERTISED_HOST_NAME=zookeeper diff --git a/integration/spring-kafka-avro/src/appTest/java/com/example/kafka/KafkaAvroApplicationCrTests.java b/integration/spring-kafka-avro/src/appTest/java/com/example/kafka/KafkaAvroApplicationCrTests.java new file mode 100644 index 0000000..61c3f59 --- /dev/null +++ b/integration/spring-kafka-avro/src/appTest/java/com/example/kafka/KafkaAvroApplicationCrTests.java @@ -0,0 +1,46 @@ +/* + * Copyright 2022 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package com.example.kafka; + +import java.time.Duration; + +import org.awaitility.Awaitility; +import org.junit.jupiter.api.Test; + +import org.springframework.cr.smoketest.support.assertj.AssertableOutput; +import org.springframework.cr.smoketest.support.junit.ApplicationTest; + +import static org.assertj.core.api.Assertions.assertThat; + +@ApplicationTest +class KafkaAvroApplicationCrTests { + + @Test + void kafkaListenerMethodReceivesMessageAndSendsResponse(AssertableOutput output) { + Awaitility.await() + .atMost(Duration.ofSeconds(30)) + .untilAsserted(() -> assertThat(output).hasSingleLineContaining("++++++1:Received Thing1:") + .hasSingleLineContaining("++++++2:Received Thing2:") + .hasSingleLineContaining("++++++3:Received Thing3:") + .hasSingleLineContaining("++++++4:Received Thing4:") + .hasSingleLineContaining("++++++5:Received Thing5:") + .hasSingleLineContaining("++++++6:Received Thing6:") + .hasSingleLineContaining("++++++7:Received Thing7:") + .hasSingleLineContaining("++++++8:Received Thing8:")); + } + +} diff --git a/integration/spring-kafka-avro/src/main/avro/Thing1.avsc b/integration/spring-kafka-avro/src/main/avro/Thing1.avsc new file mode 100644 index 0000000..975ebdd --- /dev/null +++ b/integration/spring-kafka-avro/src/main/avro/Thing1.avsc @@ -0,0 +1,10 @@ +{ +"namespace":"com.example.kafka", +"type":"record", +"name":"Thing1", +"fields":[ + {"name":"stringField", "type":"string"}, + {"name":"intField", "type":"int"} + ] +} + diff --git a/integration/spring-kafka-avro/src/main/avro/Thing2.avsc b/integration/spring-kafka-avro/src/main/avro/Thing2.avsc new file mode 100644 index 0000000..9f98ba5 --- /dev/null +++ b/integration/spring-kafka-avro/src/main/avro/Thing2.avsc @@ -0,0 +1,10 @@ +{ +"namespace":"com.example.kafka", +"type":"record", +"name":"Thing2", +"fields":[ + {"name":"stringField", "type":"string"}, + {"name":"intField", "type":"int"} + ] +} + diff --git a/integration/spring-kafka-avro/src/main/avro/Thing3.avsc b/integration/spring-kafka-avro/src/main/avro/Thing3.avsc new file mode 100644 index 0000000..65615ed --- /dev/null +++ b/integration/spring-kafka-avro/src/main/avro/Thing3.avsc @@ -0,0 +1,10 @@ +{ +"namespace":"com.example.kafka", +"type":"record", +"name":"Thing3", +"fields":[ + {"name":"stringField", "type":"string"}, + {"name":"intField", "type":"int"} + ] +} + diff --git a/integration/spring-kafka-avro/src/main/avro/Thing4.avsc b/integration/spring-kafka-avro/src/main/avro/Thing4.avsc new file mode 100644 index 0000000..ea97c08 --- /dev/null +++ b/integration/spring-kafka-avro/src/main/avro/Thing4.avsc @@ -0,0 +1,10 @@ +{ +"namespace":"com.example.kafka", +"type":"record", +"name":"Thing4", +"fields":[ + {"name":"stringField", "type":"string"}, + {"name":"intField", "type":"int"} + ] +} + diff --git a/integration/spring-kafka-avro/src/main/avro/Thing5.avsc b/integration/spring-kafka-avro/src/main/avro/Thing5.avsc new file mode 100644 index 0000000..6a5ee2c --- /dev/null +++ b/integration/spring-kafka-avro/src/main/avro/Thing5.avsc @@ -0,0 +1,10 @@ +{ +"namespace":"com.example.kafka", +"type":"record", +"name":"Thing5", +"fields":[ + {"name":"stringField", "type":"string"}, + {"name":"intField", "type":"int"} + ] +} + diff --git a/integration/spring-kafka-avro/src/main/avro/Thing6.avsc b/integration/spring-kafka-avro/src/main/avro/Thing6.avsc new file mode 100644 index 0000000..5a3ba2a --- /dev/null +++ b/integration/spring-kafka-avro/src/main/avro/Thing6.avsc @@ -0,0 +1,10 @@ +{ +"namespace":"com.example.kafka", +"type":"record", +"name":"Thing6", +"fields":[ + {"name":"stringField", "type":"string"}, + {"name":"intField", "type":"int"} + ] +} + diff --git a/integration/spring-kafka-avro/src/main/avro/Thing7.avsc b/integration/spring-kafka-avro/src/main/avro/Thing7.avsc new file mode 100644 index 0000000..673d97c --- /dev/null +++ b/integration/spring-kafka-avro/src/main/avro/Thing7.avsc @@ -0,0 +1,10 @@ +{ +"namespace":"com.example.kafka", +"type":"record", +"name":"Thing7", +"fields":[ + {"name":"stringField", "type":"string"}, + {"name":"intField", "type":"int"} + ] +} + diff --git a/integration/spring-kafka-avro/src/main/avro/Thing8.avsc b/integration/spring-kafka-avro/src/main/avro/Thing8.avsc new file mode 100644 index 0000000..b868beb --- /dev/null +++ b/integration/spring-kafka-avro/src/main/avro/Thing8.avsc @@ -0,0 +1,10 @@ +{ +"namespace":"com.example.kafka", +"type":"record", +"name":"Thing8", +"fields":[ + {"name":"stringField", "type":"string"}, + {"name":"intField", "type":"int"} + ] +} + diff --git a/integration/spring-kafka-avro/src/main/java/com/example/kafka/KafkaAvroApplication.java b/integration/spring-kafka-avro/src/main/java/com/example/kafka/KafkaAvroApplication.java new file mode 100644 index 0000000..acb20ad --- /dev/null +++ b/integration/spring-kafka-avro/src/main/java/com/example/kafka/KafkaAvroApplication.java @@ -0,0 +1,193 @@ +package com.example.kafka; + +import java.util.List; + +import org.apache.avro.specific.SpecificRecord; +import org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.clients.consumer.Consumer; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; + +import org.springframework.boot.ApplicationRunner; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.annotation.Bean; +import org.springframework.kafka.annotation.KafkaHandler; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; +import org.springframework.kafka.config.TopicBuilder; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaConsumerFactory; +import org.springframework.kafka.core.DefaultKafkaProducerFactory; +import org.springframework.kafka.core.KafkaAdmin; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.kafka.core.ProducerFactory; +import org.springframework.kafka.listener.BatchMessageListener; +import org.springframework.kafka.listener.CommonContainerStoppingErrorHandler; +import org.springframework.kafka.listener.ConcurrentMessageListenerContainer; +import org.springframework.kafka.listener.MessageListener; +import org.springframework.kafka.support.Acknowledgment; +import org.springframework.stereotype.Component; + +@SpringBootApplication +public class KafkaAvroApplication { + + public static void main(String[] args) { + System.out.println(System.getProperty("java.class.path")); + SpringApplication.run(KafkaAvroApplication.class, args); + } + + @Bean + public KafkaAdmin.NewTopics topics() { + return new KafkaAdmin.NewTopics(TopicBuilder.name("cr1").partitions(1).replicas(1).build(), + TopicBuilder.name("cr2").partitions(1).replicas(1).build(), + TopicBuilder.name("cr3").partitions(1).replicas(1).build(), + TopicBuilder.name("cr4").partitions(1).replicas(1).build(), + TopicBuilder.name("cr5").partitions(1).replicas(1).build(), + TopicBuilder.name("cr6").partitions(1).replicas(1).build(), + TopicBuilder.name("cr7").partitions(1).replicas(1).build(), + TopicBuilder.name("cr8").partitions(1).replicas(1).build()); + } + + @Bean + public ConcurrentMessageListenerContainer container3(NotAComponentMessageListener listener, + ConsumerFactory cf, ProducerFactory pf, ConcurrentKafkaListenerContainerFactory factory) { + + ((DefaultKafkaConsumerFactory) cf).setValueDeserializer(new ThingAvroSerde()); + ((DefaultKafkaProducerFactory) pf).setValueSerializer(new ThingAvroSerde()); + factory.setCommonErrorHandler(new CommonContainerStoppingErrorHandler()); + ConcurrentMessageListenerContainer container = factory.createContainer("cr3"); + container.getContainerProperties().setGroupId("cr3"); + container.getContainerProperties().setMessageListener(listener); + return container; + } + + @Bean + public ConcurrentMessageListenerContainer container6(BML6 listener, ConsumerFactory cf, + ProducerFactory pf, ConcurrentKafkaListenerContainerFactory factory) { + + factory.setCommonErrorHandler(new CommonContainerStoppingErrorHandler()); + ConcurrentMessageListenerContainer container = factory.createContainer("cr6"); + container.getContainerProperties().setGroupId("cr6"); + container.getContainerProperties().setMessageListener(listener); + return container; + } + + @Bean + public ConcurrentMessageListenerContainer container7(BML7 listener, ConsumerFactory cf, + ProducerFactory pf, ConcurrentKafkaListenerContainerFactory factory) { + + factory.setCommonErrorHandler(new CommonContainerStoppingErrorHandler()); + ConcurrentMessageListenerContainer container = factory.createContainer("cr7"); + container.getContainerProperties().setGroupId("cr7"); + container.getContainerProperties().setMessageListener(listener); + return container; + } + + @Bean + NotAComponentMessageListener otherListner() { + return new NotAComponentMessageListener(); + } + + @Bean + public ApplicationRunner runner(KafkaTemplate template) { + return args -> { + Thing1 thing1 = Thing1.newBuilder().setStringField("thing1Value").setIntField(42).build(); + template.send("cr1", thing1); + Thing2 thing2 = Thing2.newBuilder().setStringField("thing2Value").setIntField(42).build(); + template.send("cr2", thing2); + Thing3 thing3 = Thing3.newBuilder().setStringField("thing3Value").setIntField(42).build(); + template.send("cr3", thing3); + Thing4 thing4 = Thing4.newBuilder().setStringField("thing4Value").setIntField(42).build(); + template.send("cr4", thing4); + Thing5 thing5 = Thing5.newBuilder().setStringField("thing5Value").setIntField(42).build(); + template.send("cr5", thing5); + Thing6 thing6 = Thing6.newBuilder().setStringField("thing5Value").setIntField(42).build(); + template.send("cr6", thing6); + Thing7 thing7 = Thing7.newBuilder().setStringField("thing5Value").setIntField(42).build(); + template.send("cr7", thing7); + Thing8 thing8 = Thing8.newBuilder().setStringField("thing5Value").setIntField(42).build(); + template.send("cr8", thing8); + System.out.println("++++++Sent:" + thing1 + thing2 + thing3 + thing4 + thing5 + thing6 + thing7 + thing8); + }; + } + +} + +@Component +class RecordListener { + + @KafkaListener(id = "cr", topics = "cr1") + void listen1(Thing1 thing1) { + System.out.println("++++++1:Received " + thing1.getClass().getSimpleName() + ":" + thing1); + } + + @KafkaListener(id = "cr2", topics = "cr2") + void listen2(ConsumerRecord record) { + System.out.println("++++++2:Received " + record.value().getClass().getSimpleName() + ":" + record); + } + + @KafkaListener(id = "cr4", topics = "cr4", batch = "true") + void listen4(List records) { + System.out.println("++++++4:Received " + records.get(0).getClass().getSimpleName() + ":" + records.get(0)); + } + + @KafkaListener(id = "cr8", topics = "cr8", batch = "true") + void listen8(List> records) { + ConsumerRecord record = records.iterator().next(); + System.out.println("++++++8:Received " + record.value().getClass().getSimpleName() + ":" + record.value()); + } + +} + +class NotAComponentMessageListener implements MessageListener { + + @Override + public void onMessage(ConsumerRecord record) { + System.out.println("++++++3:Received " + record.value().getClass().getSimpleName() + ":" + record); + } + +} + +@Component +class BML6 implements BatchMessageListener { + + @Override + public void onMessage(List> records) { + System.out + .println("++++++6:Received " + records.get(0).value().getClass().getSimpleName() + ":" + records.get(0)); + } + +} + +@Component +class BML7 implements BatchMessageListener { + + @Override + public void onMessage(List> data) { + } + + @Override + public void onMessage(ConsumerRecords records, Acknowledgment acknowledgment, + Consumer consumer) { + ConsumerRecord record = records.iterator().next(); + System.out.println("++++++7:Received " + record.value().getClass().getSimpleName() + ":" + record); + } + + @Override + public boolean wantsPollResult() { + return true; + } + +} + +@Component +@KafkaListener(id = "cr5", topics = "cr5") +class Thing5Listener { + + @KafkaHandler + public void listen(Thing5 thing5) { + System.out.println("++++++5:Received " + thing5.getClass().getSimpleName() + ":" + thing5); + } + +} diff --git a/integration/spring-kafka-avro/src/main/java/com/example/kafka/ThingAvroSerde.java b/integration/spring-kafka-avro/src/main/java/com/example/kafka/ThingAvroSerde.java new file mode 100644 index 0000000..d0a0262 --- /dev/null +++ b/integration/spring-kafka-avro/src/main/java/com/example/kafka/ThingAvroSerde.java @@ -0,0 +1,117 @@ +package com.example.kafka; + +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.util.Arrays; +import java.util.Map; + +import org.apache.avro.generic.GenericDatumWriter; +import org.apache.avro.generic.GenericRecord; +import org.apache.avro.io.BinaryEncoder; +import org.apache.avro.io.DatumReader; +import org.apache.avro.io.DatumWriter; +import org.apache.avro.io.Decoder; +import org.apache.avro.io.DecoderFactory; +import org.apache.avro.io.EncoderFactory; +import org.apache.avro.specific.SpecificDatumReader; +import org.apache.avro.specific.SpecificRecord; +import org.apache.kafka.common.errors.SerializationException; +import org.apache.kafka.common.header.Headers; +import org.apache.kafka.common.serialization.Deserializer; +import org.apache.kafka.common.serialization.Serializer; + +public class ThingAvroSerde implements Serializer, Deserializer { + + @Override + public Object deserialize(String topic, byte[] data) { + return null; + } + + @Override + public void configure(Map configs, boolean isKey) { + Serializer.super.configure(configs, isKey); + } + + @Override + public Object deserialize(String topic, Headers headers, byte[] data) { + try { + Object result = null; + + if (data != null) { + DatumReader datumReader; + switch (new String(headers.lastHeader("thing").value())) { + case "Thing1": + datumReader = new SpecificDatumReader<>(new Thing1().getSchema()); + break; + case "Thing2": + datumReader = new SpecificDatumReader<>(new Thing2().getSchema()); + break; + case "Thing3": + datumReader = new SpecificDatumReader<>(new Thing3().getSchema()); + break; + case "Thing4": + datumReader = new SpecificDatumReader<>(new Thing4().getSchema()); + break; + case "Thing5": + datumReader = new SpecificDatumReader<>(new Thing5().getSchema()); + break; + case "Thing6": + datumReader = new SpecificDatumReader<>(new Thing6().getSchema()); + break; + case "Thing7": + datumReader = new SpecificDatumReader<>(new Thing7().getSchema()); + break; + case "Thing8": + datumReader = new SpecificDatumReader<>(new Thing8().getSchema()); + break; + default: + datumReader = null; + } + Decoder decoder = DecoderFactory.get().binaryDecoder(data, null); + + result = datumReader.read(null, decoder); + } + return result; + } + catch (Exception ex) { + throw new SerializationException( + "Can't deserialize data '" + Arrays.toString(data) + "' from topic '" + topic + "'", ex); + } + } + + @Override + public byte[] serialize(String topic, SpecificRecord data) { + return new byte[0]; + } + + @Override + public byte[] serialize(String topic, Headers headers, SpecificRecord data) { + try { + byte[] result = null; + + if (data != null) { + ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(); + BinaryEncoder binaryEncoder = EncoderFactory.get().binaryEncoder(byteArrayOutputStream, null); + + DatumWriter datumWriter = new GenericDatumWriter<>(data.getSchema()); + datumWriter.write(data, binaryEncoder); + + binaryEncoder.flush(); + byteArrayOutputStream.close(); + + result = byteArrayOutputStream.toByteArray(); + + headers.add("thing", data.getClass().getSimpleName().getBytes()); + } + return result; + } + catch (IOException ex) { + throw new SerializationException("Can't serialize data='" + data + "' for topic='" + topic + "'", ex); + } + } + + @Override + public void close() { + } + +} diff --git a/integration/spring-kafka-avro/src/main/resources/application.properties b/integration/spring-kafka-avro/src/main/resources/application.properties new file mode 100644 index 0000000..eb7491a --- /dev/null +++ b/integration/spring-kafka-avro/src/main/resources/application.properties @@ -0,0 +1,4 @@ +spring.kafka.bootstrap-servers=localhost:${KAFKA_PORT_9092:9092} + +spring.kafka.consumer.auto-offset-reset=earliest +spring.kafka.consumer.group-id=cr3