[Samples] Make sample-pulsar-binder testable

spring-pulsar-sample-apps/sample-pulsar-binder/build.gradle

@@ -22,6 +22,15 @@ ext['pulsar.version'] = "${pulsarVersion}"
 dependencies {
 	implementation "org.springframework.boot:spring-boot-starter-pulsar"
 	implementation "org.springframework.cloud:spring-cloud-stream-binder-pulsar:${springCloudStreamVersion}"
+	developmentOnly 'org.springframework.boot:spring-boot-docker-compose'
+
+	testImplementation project(':spring-pulsar-test')
+	testRuntimeOnly 'ch.qos.logback:logback-classic'
+	testImplementation "org.springframework.boot:spring-boot-starter-test"
+	testImplementation "org.springframework.boot:spring-boot-testcontainers"
+	testImplementation 'org.testcontainers:junit-jupiter'
+	testImplementation 'org.testcontainers:pulsar'
+
 }
 
 test {
@@ -36,4 +45,6 @@ bootRun {
 			"--add-opens", "java.base/java.util=ALL-UNNAMED",
 			"--add-opens", "java.base/sun.net=ALL-UNNAMED"
 	]
+	// when run from command line, path must be set relative to module dir
+	systemProperty 'spring.docker.compose.file', 'compose.yaml'
 }

spring-pulsar-sample-apps/sample-pulsar-binder/compose.yaml

@@ -0,0 +1,7 @@
+services:
+  pulsar:
+    image: 'apachepulsar/pulsar:3.1.2'
+    ports:
+      - '6650'
+      - '8080'
+    command: 'bin/pulsar standalone'

spring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/com/example/SpringPulsarBinderSampleApp.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2022-2024 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;
+
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.springframework.boot.ApplicationRunner;
+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 SpringPulsarBinderSampleApp {
+
+	private static final Logger LOG = LoggerFactory.getLogger(SpringPulsarBinderSampleApp.class);
+
+	public static void main(String[] args) {
+		SpringApplication.run(SpringPulsarBinderSampleApp.class, args);
+	}
+
+	private AtomicInteger counter = new AtomicInteger();
+
+	@Bean
+	ApplicationRunner fooSupplier(StreamBridge streamBridge) {
+		return (args) -> {
+			for (int i = 0; i < 10; i++) {
+				var foo = new Foo("fooSupplier:" + i);
+				streamBridge.send("fooSupplier-out-0", foo);
+				LOG.info("++++++SOURCE {}------", foo);
+			}
+		};
+	}
+
+	@Bean
+	public Function<Foo, Bar> fooProcessor() {
+		return (foo) -> {
+			var bar = new Bar(foo);
+			LOG.info("++++++PROCESSOR {} --> {}------", foo, bar);
+			return bar;
+		};
+	}
+
+	@Bean
+	public Consumer<Bar> barLogger() {
+		return (bar) -> LOG.info("++++++SINK {}------", bar);
+	}
+
+	record Foo(String value) {
+	}
+
+	record Bar(Foo value) {
+	}
+
+}

spring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/com/example/package-info.java

@@ -3,7 +3,7 @@
  */
 @NonNullApi
 @NonNullFields
-package org.springframework.pulsar.sample.binder;
+package com.example;
 
 import org.springframework.lang.NonNullApi;
 import org.springframework.lang.NonNullFields;

spring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/org/springframework/pulsar/sample/binder/SpringPulsarBinderSampleApp.java

@@ -1,104 +0,0 @@
-/*
- * 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 org.springframework.pulsar.sample.binder;
-
-import java.util.function.Consumer;
-import java.util.function.Function;
-import java.util.function.Supplier;
-
-import org.apache.pulsar.common.schema.SchemaType;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import org.springframework.boot.SpringApplication;
-import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.context.annotation.Bean;
-import org.springframework.pulsar.annotation.PulsarListener;
-
-/**
- * This sample binder app has an extra consumer that is equipped with Pulsar's DLT feature
- * - timeLoggerToDlt. However, this consumer is not part of the
- * spring.cloud.function.definition. In order to enable this, add the function
- * timeLoggerToDlt to the definition in the application.yml file. When doing this, in
- * order to minimize verbose output and just to focus on the DLT feature, comment out the
- * regular supplier below (timeSupplier) and then un-comment the ApplicationRunner below.
- * The runner only sends a single message whereas the supplier sends a message every
- * second.
- *
- * @author Soby Chacko
- */
-@SpringBootApplication
-public class SpringPulsarBinderSampleApp {
-
-	private final Logger logger = LoggerFactory.getLogger(this.getClass());
-
-	public static void main(String[] args) {
-		SpringApplication.run(SpringPulsarBinderSampleApp.class, args);
-	}
-
-	@Bean
-	public Supplier<Time> timeSupplier() {
-		return () -> new Time(String.valueOf(System.currentTimeMillis()));
-	}
-
-	// @Bean
-	// ApplicationRunner runner(PulsarTemplate<Time> pulsarTemplate) {
-	//
-	// String topic = "timeSupplier-out-0";
-	//
-	// return args -> {
-	// for (int i = 0; i < 1; i++) {
-	// pulsarTemplate.send(topic, new Time(String.valueOf(System.currentTimeMillis())),
-	// Schema.JSON(Time.class));
-	// }
-	// };
-	// }
-
-	@Bean
-	public Function<Time, EnhancedTime> timeProcessor() {
-		return (time) -> {
-			EnhancedTime enhancedTime = new EnhancedTime(time, "5150");
-			this.logger.info("PROCESSOR: {} --> {}", time, enhancedTime);
-			return enhancedTime;
-		};
-	}
-
-	@Bean
-	public Consumer<EnhancedTime> timeLogger() {
-		return (time) -> this.logger.info("SINK:      {}", time);
-	}
-
-	@Bean
-	public Consumer<EnhancedTime> timeLoggerToDlt() {
-		return (time) -> {
-			this.logger.info("SINK (TO DLT EVENTUALLY):      {}", time);
-			throw new RuntimeException("fail " + time);
-		};
-	}
-
-	@PulsarListener(id = "dlqListener", topics = "notification-dlq", schemaType = SchemaType.JSON)
-	void listenDlq(EnhancedTime msg) {
-		this.logger.info("From DLQ: {}", msg);
-	}
-
-	record Time(String time) {
-	}
-
-	record EnhancedTime(Time time, String extra) {
-	}
-
-}

spring-pulsar-sample-apps/sample-pulsar-binder/src/main/resources/application.yml

@@ -1,52 +1,44 @@
 spring:
   cloud:
     function:
-      definition: timeSupplier;timeProcessor;timeLogger
+      definition: fooProcessor;barLogger
     stream:
+      output-bindings: fooSupplier-out-0
       bindings:
-        timeSupplier-out-0:
+        fooSupplier-out-0:
           producer:
             use-native-encoding: true
-        timeProcessor-in-0:
-          destination: timeSupplier-out-0
+        fooProcessor-in-0:
+          destination: fooSupplier-out-0
           consumer:
             use-native-decoding: true
-        timeProcessor-out-0:
-          destination: timeProcessor-out-0
+        fooProcessor-out-0:
+          destination: fooProcessor-out-0
           producer:
             use-native-encoding: true
-        timeLogger-in-0:
-          destination: timeProcessor-out-0
-          consumer:
-            use-native-decoding: true
-        timeLoggerToDlt-in-0:
-          destination: timeProcessor-out-0
+        barLogger-in-0:
+          destination: fooProcessor-out-0
           consumer:
             use-native-decoding: true
       pulsar:
         bindings:
-          timeSupplier-out-0:
+          fooSupplier-out-0:
             producer:
               schema-type: JSON
-              message-type: org.springframework.pulsar.sample.binder.SpringPulsarBinderSampleApp.Time
-          timeProcessor-in-0:
+              message-type: com.example.SpringPulsarBinderSampleApp.Foo
+          fooProcessor-in-0:
             consumer:
               schema-type: JSON
-              message-type: org.springframework.pulsar.sample.binder.SpringPulsarBinderSampleApp.Time
-          timeProcessor-out-0:
+              message-type: com.example.SpringPulsarBinderSampleApp.Foo
+          fooProcessor-out-0:
             producer:
               schema-type: JSON
-              message-type: org.springframework.pulsar.sample.binder.SpringPulsarBinderSampleApp.EnhancedTime
-          timeLogger-in-0:
+              message-type: com.example.SpringPulsarBinderSampleApp.Bar
+          barLogger-in-0:
             consumer:
               schema-type: JSON
-              message-type: org.springframework.pulsar.sample.binder.SpringPulsarBinderSampleApp.EnhancedTime
-          timeLoggerToDlt-in-0:
-            consumer:
-              subscription-type: Shared
-              negative-ack-redelivery-delay: 1s
-              dead-letter-policy:
-                dead-letter-topic: notification-dlq
-                max-redeliver-count: 5
-              schema-type: JSON
-              message-type: org.springframework.pulsar.sample.binder.SpringPulsarBinderSampleApp.EnhancedTime
+              message-type: com.example.SpringPulsarBinderSampleApp.Bar
+  docker:
+    compose:
+      # when run from Intellij via "Run" button, path must be set from project root
+      file: spring-pulsar-sample-apps/sample-pulsar-binder/compose.yaml

spring-pulsar-sample-apps/sample-pulsar-binder/src/test/java/com/example/SpringPulsarBinderSampleAppTests.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2012-2024 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;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.stream.IntStream;
+
+import com.example.SpringPulsarBinderSampleApp.Bar;
+import com.example.SpringPulsarBinderSampleApp.Foo;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.boot.test.system.CapturedOutput;
+import org.springframework.boot.test.system.OutputCaptureExtension;
+import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
+import org.springframework.test.context.DynamicPropertyRegistry;
+import org.springframework.test.context.DynamicPropertySource;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@SpringBootTest
+@ExtendWith(OutputCaptureExtension.class)
+class SpringPulsarBinderSampleAppTests implements PulsarTestContainerSupport {
+
+	@DynamicPropertySource
+	static void pulsarProperties(DynamicPropertyRegistry registry) {
+		registry.add("spring.pulsar.client.service-url", PULSAR_CONTAINER::getPulsarBrokerUrl);
+		registry.add("spring.pulsar.admin.service-url", PULSAR_CONTAINER::getHttpServiceUrl);
+	}
+
+	@Test
+	void produceConsumeWithPrimitiveMessageType(CapturedOutput output) {
+		var expectedOutput = new ArrayList<String>();
+		IntStream.range(0, 10).forEachOrdered((i) -> {
+			var foo = new Foo("fooSupplier:" + i);
+			var bar = new Bar(foo);
+			expectedOutput.add("++++++SOURCE %s------".formatted(foo));
+			expectedOutput.add("++++++PROCESSOR %s --> %s------".formatted(foo, bar));
+			expectedOutput.add("++++++SINK %s------".formatted(bar));
+		});
+		Awaitility.waitAtMost(Duration.ofSeconds(15))
+				.untilAsserted(() -> assertThat(output).contains(expectedOutput));
+	}
+
+}

spring-pulsar-sample-apps/sample-pulsar-binder/src/test/resources/logback-test.xml

@@ -0,0 +1,14 @@
+<configuration>
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger - %msg%n</pattern>
+        </encoder>
+    </appender>
+    <root level="WARN">
+        <appender-ref ref="STDOUT"/>
+    </root>
+    <logger name="com.example" level="INFO"/>
+    <logger name="com.github.dockerjava" level="ERROR"/>
+    <logger name="org.apache.pulsar.common.util.netty" level="ERROR" />
+    <logger name="org.testcontainers" level="ERROR"/>
+</configuration>