Add smoke test for Spring Kafka with AVRO

Closes gh-11

integration/spring-kafka-avro/README.adoc

@@ -0,0 +1 @@
+Verifies that Spring Kafka stream with AVRO encoding is CR compatible.

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") }
+}

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

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:"));
+	}
+
+}

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"}
+	]
+}
+

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"}
+	]
+}
+

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"}
+	]
+}
+

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"}
+	]
+}
+

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"}
+	]
+}
+

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"}
+	]
+}
+

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"}
+	]
+}
+

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"}
+	]
+}
+

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<Object, Object> container3(NotAComponentMessageListener listener,
+			ConsumerFactory cf, ProducerFactory pf, ConcurrentKafkaListenerContainerFactory<Object, Object> factory) {
+
+		((DefaultKafkaConsumerFactory<Object, Object>) cf).setValueDeserializer(new ThingAvroSerde());
+		((DefaultKafkaProducerFactory<Object, SpecificRecord>) pf).setValueSerializer(new ThingAvroSerde());
+		factory.setCommonErrorHandler(new CommonContainerStoppingErrorHandler());
+		ConcurrentMessageListenerContainer<Object, Object> container = factory.createContainer("cr3");
+		container.getContainerProperties().setGroupId("cr3");
+		container.getContainerProperties().setMessageListener(listener);
+		return container;
+	}
+
+	@Bean
+	public ConcurrentMessageListenerContainer<Object, Object> container6(BML6 listener, ConsumerFactory cf,
+			ProducerFactory pf, ConcurrentKafkaListenerContainerFactory<Object, Object> factory) {
+
+		factory.setCommonErrorHandler(new CommonContainerStoppingErrorHandler());
+		ConcurrentMessageListenerContainer<Object, Object> container = factory.createContainer("cr6");
+		container.getContainerProperties().setGroupId("cr6");
+		container.getContainerProperties().setMessageListener(listener);
+		return container;
+	}
+
+	@Bean
+	public ConcurrentMessageListenerContainer<Object, Object> container7(BML7 listener, ConsumerFactory cf,
+			ProducerFactory pf, ConcurrentKafkaListenerContainerFactory<Object, Object> factory) {
+
+		factory.setCommonErrorHandler(new CommonContainerStoppingErrorHandler());
+		ConcurrentMessageListenerContainer<Object, Object> 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<Object, Object> 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<String, Thing2> record) {
+		System.out.println("++++++2:Received " + record.value().getClass().getSimpleName() + ":" + record);
+	}
+
+	@KafkaListener(id = "cr4", topics = "cr4", batch = "true")
+	void listen4(List<Thing4> records) {
+		System.out.println("++++++4:Received " + records.get(0).getClass().getSimpleName() + ":" + records.get(0));
+	}
+
+	@KafkaListener(id = "cr8", topics = "cr8", batch = "true")
+	void listen8(List<ConsumerRecord<String, Thing8>> records) {
+		ConsumerRecord<String, Thing8> record = records.iterator().next();
+		System.out.println("++++++8:Received " + record.value().getClass().getSimpleName() + ":" + record.value());
+	}
+
+}
+
+class NotAComponentMessageListener implements MessageListener<String, Thing3> {
+
+	@Override
+	public void onMessage(ConsumerRecord<String, Thing3> record) {
+		System.out.println("++++++3:Received " + record.value().getClass().getSimpleName() + ":" + record);
+	}
+
+}
+
+@Component
+class BML6 implements BatchMessageListener<String, Thing6> {
+
+	@Override
+	public void onMessage(List<ConsumerRecord<String, Thing6>> records) {
+		System.out
+			.println("++++++6:Received " + records.get(0).value().getClass().getSimpleName() + ":" + records.get(0));
+	}
+
+}
+
+@Component
+class BML7 implements BatchMessageListener<String, Thing7> {
+
+	@Override
+	public void onMessage(List<ConsumerRecord<String, Thing7>> data) {
+	}
+
+	@Override
+	public void onMessage(ConsumerRecords<String, Thing7> records, Acknowledgment acknowledgment,
+			Consumer<String, Thing7> consumer) {
+		ConsumerRecord<String, Thing7> 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);
+	}
+
+}

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<SpecificRecord>, Deserializer<Object> {
+
+	@Override
+	public Object deserialize(String topic, byte[] data) {
+		return null;
+	}
+
+	@Override
+	public void configure(Map<String, ?> 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<GenericRecord> 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<SpecificRecord> 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() {
+	}
+
+}

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