diff --git a/integration/spring-kafka-streams/README.adoc b/integration/spring-kafka-streams/README.adoc new file mode 100644 index 0000000..2872ce5 --- /dev/null +++ b/integration/spring-kafka-streams/README.adoc @@ -0,0 +1,3 @@ +Tests if Spring Kafka pipeline is CR compatible. + + diff --git a/integration/spring-kafka-streams/build.gradle b/integration/spring-kafka-streams/build.gradle new file mode 100644 index 0000000..28376ef --- /dev/null +++ b/integration/spring-kafka-streams/build.gradle @@ -0,0 +1,21 @@ +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") + implementation("org.springframework.kafka:spring-kafka:3.0.10-SNAPSHOT") // TODO: remove the version once 3.0.10+ is released. + implementation("org.apache.kafka:kafka-streams") + 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-streams/docker-compose.yml b/integration/spring-kafka-streams/docker-compose.yml new file mode 100644 index 0000000..5850c55 --- /dev/null +++ b/integration/spring-kafka-streams/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-streams/src/appTest/java/com/example/kafka/KafkaStreamsApplicationCrTests.java b/integration/spring-kafka-streams/src/appTest/java/com/example/kafka/KafkaStreamsApplicationCrTests.java new file mode 100644 index 0000000..3414f5c --- /dev/null +++ b/integration/spring-kafka-streams/src/appTest/java/com/example/kafka/KafkaStreamsApplicationCrTests.java @@ -0,0 +1,38 @@ +/* + * Copyright 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 KafkaStreamsApplicationCrTests { + + @Test + void kafkaListenerMethodReceivesMessageAndSendsResponse(AssertableOutput output) { + Awaitility.await() + .atMost(Duration.ofSeconds(30)) + .untilAsserted(() -> assertThat(output).hasSingleLineContaining("++++++Received:FOO")); + } + +} diff --git a/integration/spring-kafka-streams/src/main/java/com/example/kafka/KafkaStreamsApplication.java b/integration/spring-kafka-streams/src/main/java/com/example/kafka/KafkaStreamsApplication.java new file mode 100644 index 0000000..473c8a5 --- /dev/null +++ b/integration/spring-kafka-streams/src/main/java/com/example/kafka/KafkaStreamsApplication.java @@ -0,0 +1,70 @@ +/* + * Copyright 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 org.apache.kafka.clients.admin.NewTopic; +import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.StreamsBuilder; +import org.apache.kafka.streams.kstream.KStream; + +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.EnableKafkaStreams; +import org.springframework.kafka.annotation.KafkaListener; +import org.springframework.kafka.config.TopicBuilder; +import org.springframework.kafka.core.KafkaTemplate; + +@SpringBootApplication +@EnableKafkaStreams +public class KafkaStreamsApplication { + + public static void main(String[] args) { + SpringApplication.run(KafkaStreamsApplication.class, args); + } + + @Bean + KStream stream(StreamsBuilder builder) { + KStream stream = builder.stream("crStreamsIn"); + stream.map((k, v) -> new KeyValue<>(k, v.toUpperCase())).to("crStreamsOut"); + return stream; + } + + @KafkaListener(id = "cr", topics = "crStreamsOut") + public void listen(String in) { + System.out.println("++++++Received:" + in); + } + + @Bean + public NewTopic topic1() { + return TopicBuilder.name("crStreamsIn").partitions(1).replicas(1).build(); + } + + @Bean + public NewTopic topic2() { + return TopicBuilder.name("crStreamsOut").partitions(1).replicas(1).build(); + } + + @Bean + public ApplicationRunner runner(KafkaTemplate template) { + return args -> { + template.send("crStreamsIn", "foo"); + System.out.println("++++++Sent:foo"); + }; + } + +} diff --git a/integration/spring-kafka-streams/src/main/resources/application.properties b/integration/spring-kafka-streams/src/main/resources/application.properties new file mode 100644 index 0000000..b321271 --- /dev/null +++ b/integration/spring-kafka-streams/src/main/resources/application.properties @@ -0,0 +1,7 @@ +spring.application.name=crKafkaStreams + +spring.kafka.bootstrap-servers=localhost:${KAFKA_PORT_9092:9092} +spring.kafka.consumer.auto-offset-reset=earliest + +spring.kafka.streams.properties.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde +spring.kafka.streams.properties.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde