Add Spring Cloud Stream Pulsar binder test

See gh-172

README.adoc

@@ -53,7 +53,7 @@ for example
 === Add a new smoke test
 
 1. Create a new directory for your smoke test in the appropriate group
-2. Include the directory in `settings.gradle`
+2. Include the directory in `settings.gradle` (new groups only)
 3. Run `./gradlew updateInfrastructure` to add the smoke test to the status page and CI pipeline
 
 === Test against local changes

STATUS.adoc

@@ -258,6 +258,12 @@ h|nativeTest
 |
 |
 
+|cloud-stream-pulsar
+|image:https://ci.spring.io/api/v1/teams/spring-aot-smoke-tests/pipelines/spring-aot-smoke-tests-1.0.x/jobs/cloud-stream-pulsar-app-test/badge[link=https://ci.spring.io/teams/spring-aot-smoke-tests/pipelines/spring-aot-smoke-tests-1.0.x/jobs/cloud-stream-pulsar-app-test]
+|image:https://ci.spring.io/api/v1/teams/spring-aot-smoke-tests/pipelines/spring-aot-smoke-tests-1.0.x/jobs/cloud-stream-pulsar-native-app-test/badge[link=https://ci.spring.io/teams/spring-aot-smoke-tests/pipelines/spring-aot-smoke-tests-1.0.x/jobs/cloud-stream-pulsar-native-app-test]
+|
+|
+
 |cloud-stream-rabbit
 |image:https://ci.spring.io/api/v1/teams/spring-aot-smoke-tests/pipelines/spring-aot-smoke-tests-1.0.x/jobs/cloud-stream-rabbit-app-test/badge[link=https://ci.spring.io/teams/spring-aot-smoke-tests/pipelines/spring-aot-smoke-tests-1.0.x/jobs/cloud-stream-rabbit-app-test]
 |image:https://ci.spring.io/api/v1/teams/spring-aot-smoke-tests/pipelines/spring-aot-smoke-tests-1.0.x/jobs/cloud-stream-rabbit-native-app-test/badge[link=https://ci.spring.io/teams/spring-aot-smoke-tests/pipelines/spring-aot-smoke-tests-1.0.x/jobs/cloud-stream-rabbit-native-app-test]
@@ -799,4 +805,3 @@ h|nativeTest
 |
 
 |===
-

ci/smoke-tests.yml

@@ -116,6 +116,9 @@ groups:
      - name: cloud-stream-kafka
        app_test: true
        test: false
+     - name: cloud-stream-pulsar
+       app_test: true
+       test: false
      - name: cloud-stream-rabbit
        app_test: true
        test: false

cloud/cloud-stream-pulsar/README.adoc

@@ -0,0 +1 @@
+Tests if Spring Cloud Stream with Pulsar is working.

cloud/cloud-stream-pulsar/build.gradle

@@ -0,0 +1,25 @@
+plugins {
+	id 'java'
+	id 'org.springframework.boot'
+	id 'org.springframework.aot.smoke-test'
+	id 'org.graalvm.buildtools.native'
+}
+
+ext {
+	set('springCloudVersion', "2022.0.2-SNAPSHOT")
+	set('springPulsarVersion', "0.1.1-SNAPSHOT")
+}
+
+dependencies {
+	implementation(platform(org.springframework.boot.gradle.plugin.SpringBootPlugin.BOM_COORDINATES))
+	implementation(platform("org.springframework.cloud:spring-cloud-dependencies:${springCloudVersion}"))
+	implementation("org.springframework.boot:spring-boot-starter")
+	implementation("org.springframework.cloud:spring-cloud-stream")
+	implementation("org.springframework.pulsar:spring-pulsar-spring-boot-starter:${springPulsarVersion}")
+	implementation("org.springframework.pulsar:spring-pulsar-spring-cloud-stream-binder:${springPulsarVersion}")
+
+	testImplementation("org.springframework.boot:spring-boot-starter-test")
+
+	appTestImplementation(project(":aot-smoke-test-support"))
+	appTestImplementation("org.awaitility:awaitility:4.2.0")
+}

cloud/cloud-stream-pulsar/docker-compose.yml

@@ -0,0 +1,10 @@
+version: '3'
+services:
+  pulsar:
+    image: apachepulsar/pulsar:2.11.0
+    ports:
+      - '8080'
+      - '6650'
+    command: bin/pulsar standalone
+    healthcheck:
+      test: curl http://127.0.0.1:8080/admin/v2/namespaces/public/default

cloud/cloud-stream-pulsar/src/appTest/java/com/example/cloud/stream/pulsar/SpringCloudStreamPulsarApplicationAotTests.java

@@ -0,0 +1,34 @@
+package com.example.cloud.stream.pulsar;
+
+import java.time.Duration;
+
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.aot.smoketest.support.assertj.AssertableOutput;
+import org.springframework.aot.smoketest.support.junit.ApplicationTest;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@ApplicationTest
+class SpringCloudStreamPulsarApplicationAotTests {
+
+	@Test
+	void suppliedMessageIsUppercasedAndLogged(AssertableOutput output) {
+		// INPUT -> How much wood could a woodchuck chuck if a woodchuck could chuck
+		// wood?"
+		Awaitility.await()
+			.atMost(Duration.ofSeconds(30))
+			.untilAsserted(() -> assertThat(output).hasLineContaining("++++++Received:HOW")
+				.hasLineContaining("++++++Received:MUCH")
+				.hasLineContaining("++++++Received:WOOD")
+				.hasLineContaining("++++++Received:COULD")
+				.hasLineContaining("++++++Received:A")
+				.hasLineContaining("++++++Received:WOODCHUCK")
+				.hasLineContaining("++++++Received:CHUCK")
+				.hasLineContaining("++++++Received:IF")
+				.hasLineContaining("++++++Received:COULD")
+				.hasLineContaining("++++++Received:WOOD?"));
+	}
+
+}

cloud/cloud-stream-pulsar/src/main/java/com/example/cloud/stream/pulsar/SpringCloudStreamPulsarApplication.java

@@ -0,0 +1,50 @@
+package com.example.cloud.stream.pulsar;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.cloud.stream.function.StreamBridge;
+import org.springframework.context.annotation.Bean;
+
+@SpringBootApplication
+public class SpringCloudStreamPulsarApplication {
+
+	public static void main(String[] args) throws Exception {
+		SpringApplication.run(SpringCloudStreamPulsarApplication.class, args);
+		Thread.currentThread().join(); // To be able to measure memory consumption
+	}
+
+	@Bean
+	Supplier<Word> graalSupplier() {
+		var splitWoodchuck = "How much wood could a woodchuck chuck if a woodchuck could chuck wood?".split(" ");
+		final var wordsIndex = new AtomicInteger(0);
+		return () -> {
+			var wordIndex = wordsIndex.getAndAccumulate(splitWoodchuck.length,
+					(curIndex, numWords) -> curIndex < numWords - 1 ? curIndex + 1 : 0);
+			var strWord = splitWoodchuck[wordIndex];
+			return new Word(strWord.length(), strWord);
+		};
+	}
+
+	@Bean
+	Function<Word, String> graalUppercaseFunction() {
+		return word -> "{\"size\":%d,\"word\":\"%s\"}".formatted(word.size(), word.word().toUpperCase());
+	}
+
+	@Bean
+	public Consumer<Word> graalLoggingConsumer(StreamBridge streamBridge) {
+		return word -> {
+			System.out.println("++++++Received:" + word.word());
+			// Verifying that StreamBridge API works in native applications.
+			streamBridge.send("sb-out", word);
+		};
+	}
+
+	public record Word(int size, String word) {
+	}
+
+}

cloud/cloud-stream-pulsar/src/main/resources/application.yml

@@ -0,0 +1,23 @@
+spring.main.web-application-type: none
+
+spring.pulsar:
+  client:
+    service-url: "pulsar://${PULSAR_HOST:localhost}:${PULSAR_PORT_6650:6650}"
+  administration:
+    service-url: "http://${PULSAR_HOST:localhost}:${PULSAR_PORT_8080:8080}"
+
+spring.cloud:
+  function:
+    definition: graalSupplier;graalUppercaseFunction;graalLoggingConsumer
+  stream:
+    bindings:
+      graalSupplier-out-0:
+        destination: graalUppercaseFunction-in-0
+      graalLoggingConsumer-in-0:
+        destination: graalUppercaseFunction-out-0
+    pulsar:
+      bindings:
+        graalSupplier-out-0:
+          producer:
+            producer-name: graalSupplierOut
+# NOTE: The spring.cloud.stream.pulsar.bindings exercises ext binding props