From 52303e2935b42123789ee1c2c1212972fd5c12a8 Mon Sep 17 00:00:00 2001 From: Christian Tzolov Date: Tue, 1 Aug 2023 12:35:31 +0200 Subject: [PATCH] Add smoke test for Spring Kafka - Use a CRaC compatible spring-kafka version ( >=3.0.10-SNAPSHOT) Closes gh-9 --- integration/spring-kafka/build.gradle | 24 +++++++ integration/spring-kafka/docker-compose.yml | 25 +++++++ .../example/kafka/KafkaApplicationTests.java | 40 +++++++++++ .../com/example/kafka/KafkaApplication.java | 66 +++++++++++++++++++ .../src/main/resources/application.properties | 5 ++ 5 files changed, 160 insertions(+) create mode 100644 integration/spring-kafka/build.gradle create mode 100644 integration/spring-kafka/docker-compose.yml create mode 100644 integration/spring-kafka/src/appTest/java/com/example/kafka/KafkaApplicationTests.java create mode 100644 integration/spring-kafka/src/main/java/com/example/kafka/KafkaApplication.java create mode 100644 integration/spring-kafka/src/main/resources/application.properties diff --git a/integration/spring-kafka/build.gradle b/integration/spring-kafka/build.gradle new file mode 100644 index 0000000..246652e --- /dev/null +++ b/integration/spring-kafka/build.gradle @@ -0,0 +1,24 @@ +plugins { + id "java" + id "org.springframework.boot" + id "org.springframework.cr.smoke-test" +} + +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.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") +} diff --git a/integration/spring-kafka/docker-compose.yml b/integration/spring-kafka/docker-compose.yml new file mode 100644 index 0000000..19ecd82 --- /dev/null +++ b/integration/spring-kafka/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 \ No newline at end of file diff --git a/integration/spring-kafka/src/appTest/java/com/example/kafka/KafkaApplicationTests.java b/integration/spring-kafka/src/appTest/java/com/example/kafka/KafkaApplicationTests.java new file mode 100644 index 0000000..1b4b385 --- /dev/null +++ b/integration/spring-kafka/src/appTest/java/com/example/kafka/KafkaApplicationTests.java @@ -0,0 +1,40 @@ +/* + * Copyright 2022-2023 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 KafkaApplicationTests { + + @Test + void kafkaListenerMethodReceivesMessageAndSendsResponse(AssertableOutput output) { + Awaitility.await() + .atMost(Duration.ofSeconds(30)) + .untilAsserted(() -> assertThat(output).hasSingleLineContaining( + "++++++Received: Greeting{message='Hello from Coordinated Restore at Checkpoint!'}")); + } + +} \ No newline at end of file diff --git a/integration/spring-kafka/src/main/java/com/example/kafka/KafkaApplication.java b/integration/spring-kafka/src/main/java/com/example/kafka/KafkaApplication.java new file mode 100644 index 0000000..36b830a --- /dev/null +++ b/integration/spring-kafka/src/main/java/com/example/kafka/KafkaApplication.java @@ -0,0 +1,66 @@ +package com.example.kafka; + +import org.apache.kafka.clients.admin.NewTopic; + +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.KafkaListener; +import org.springframework.kafka.config.TopicBuilder; +import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.core.KafkaTemplate; + +import com.fasterxml.jackson.annotation.JsonCreator; +import com.fasterxml.jackson.annotation.JsonProperty; + +@SpringBootApplication +public class KafkaApplication { + + public static void main(String[] args) { + SpringApplication.run(KafkaApplication.class, args); + } + + @KafkaListener(id = "crac", topics = "crac") + public void listen(Greeting in) { + System.out.println("++++++Received: " + in); + } + + @Bean + public NewTopic topic() { + return TopicBuilder.name("crac").partitions(1).replicas(1).build(); + } + + @Bean + public ApplicationRunner runner(KafkaTemplate template, ConsumerFactory cf) { + + cf.addListener(new ConsumerFactory.Listener<>() { + }); + return args -> { + Greeting data = new Greeting("Hello from Coordinated Restore at Checkpoint!"); + template.send("crac", data); + System.out.println("++++++Sent: " + data); + }; + } + + public static class Greeting { + + private final String message; + + @JsonCreator + public Greeting(@JsonProperty("message") String message) { + this.message = message; + } + + public String getMessage() { + return this.message; + } + + @Override + public String toString() { + return "Greeting{" + "message='" + this.message + '\'' + '}'; + } + + } + +} \ No newline at end of file diff --git a/integration/spring-kafka/src/main/resources/application.properties b/integration/spring-kafka/src/main/resources/application.properties new file mode 100644 index 0000000..f24cf73 --- /dev/null +++ b/integration/spring-kafka/src/main/resources/application.properties @@ -0,0 +1,5 @@ +spring.kafka.bootstrap-servers=localhost:${KAFKA_PORT_9092:9092} +spring.kafka.consumer.auto-offset-reset=earliest +spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer +spring.kafka.consumer.properties[spring.json.trusted.packages]=com.example.kafka +spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer \ No newline at end of file