From d4e816e52ad98902292fcff8257f71e2252bcc40 Mon Sep 17 00:00:00 2001
From: Soby Chacko <chackos@vmware.com>
Date: Tue, 21 Mar 2023 18:34:10 -0400
Subject: [PATCH] Binder sample app update with DLT

---
 .../binder/SpringPulsarBinderSampleApp.java   | 38 +++++++++++++++++++
 .../src/main/resources/application.yml        | 15 +++++++-
 2 files changed, 52 insertions(+), 1 deletion(-)

diff --git a/spring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/org/springframework/pulsar/sample/binder/SpringPulsarBinderSampleApp.java b/spring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/org/springframework/pulsar/sample/binder/SpringPulsarBinderSampleApp.java
index e79d2794..d7c4b66a 100644
--- a/spring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/org/springframework/pulsar/sample/binder/SpringPulsarBinderSampleApp.java
+++ b/spring-pulsar-sample-apps/sample-pulsar-binder/src/main/java/org/springframework/pulsar/sample/binder/SpringPulsarBinderSampleApp.java
@@ -20,13 +20,26 @@ import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.function.Supplier;
 
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.common.schema.SchemaType;
 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.context.annotation.Bean;
+import org.springframework.pulsar.annotation.PulsarListener;
+import org.springframework.pulsar.core.PulsarTemplate;
 
+/**
+ * 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.
+ */
 @SpringBootApplication
 public class SpringPulsarBinderSampleApp {
 
@@ -41,6 +54,18 @@ public class SpringPulsarBinderSampleApp {
 		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) -> {
@@ -55,6 +80,19 @@ public class SpringPulsarBinderSampleApp {
 		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) {
+		System.out.println("From DLQ: " + msg);
+	}
+
 	record Time(String time) {
 	}
 
diff --git a/spring-pulsar-sample-apps/sample-pulsar-binder/src/main/resources/application.yml b/spring-pulsar-sample-apps/sample-pulsar-binder/src/main/resources/application.yml
index 8608a8b3..665014f6 100644
--- a/spring-pulsar-sample-apps/sample-pulsar-binder/src/main/resources/application.yml
+++ b/spring-pulsar-sample-apps/sample-pulsar-binder/src/main/resources/application.yml
@@ -1,7 +1,7 @@
 spring:
   cloud:
     function:
-      definition: timeSupplier;timeProcessor;timeLogger;
+      definition: timeSupplier;timeProcessor;timeLogger
     stream:
       bindings:
         timeSupplier-out-0:
@@ -19,6 +19,10 @@ spring:
           destination: timeProcessor-out-0
           consumer:
             use-native-decoding: true
+        timeLoggerToDlt-in-0:
+          destination: timeProcessor-out-0
+          consumer:
+            use-native-decoding: true
       pulsar:
         bindings:
           timeSupplier-out-0:
@@ -37,3 +41,12 @@ spring:
             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