fix: CastClassException in PubSubDeadLetterTopicSampleAppIntegrationT…

…est #3139 (#3145)

* fix: Added object mapper bean to fix issue with spring cloud stream 4.1.3 failing to deserialize objects (spring-cloud/spring-cloud-stream#2977) causing pubsub steam dead letter and functional sample test to fail

spring-cloud-gcp-samples/spring-cloud-gcp-pubsub-stream-dead-letter-sample/src/main/java/com/example/SinkExample.java

@@ -16,12 +16,15 @@
 
 package com.example;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.cloud.pubsub.v1.Subscriber;
 import com.google.cloud.spring.pubsub.support.BasicAcknowledgeablePubsubMessage;
 import com.google.cloud.spring.pubsub.support.GcpPubSubHeaders;
 import java.util.function.Consumer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.springframework.cloud.function.context.config.JsonMessageConverter;
+import org.springframework.cloud.function.json.JacksonMapper;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.messaging.Message;
@@ -63,4 +66,12 @@ public Consumer<UserMessage> deadLetterMessages() {
             userMessage.getUsername(),
             userMessage.getBody());
   }
+
+  // This is workaround for https://github.com/GoogleCloudPlatform/spring-cloud-gcp/issues/3139
+  // and should be removed once fix https://github.com/spring-cloud/spring-cloud-function/pull/1162
+  // is deployed
+  @Bean
+  public JsonMessageConverter customJsonMessageConverter(ObjectMapper objectMapper) {
+    return new JsonMessageConverter(new JacksonMapper(objectMapper));
+  }
 }

spring-cloud-gcp-samples/spring-cloud-gcp-pubsub-stream-functional-sample/spring-cloud-gcp-pubsub-stream-functional-sample-sink/src/main/java/com/example/Sink.java

@@ -17,9 +17,12 @@
 package com.example;
 
 import com.example.model.UserMessage;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import java.util.function.Consumer;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.springframework.cloud.function.context.config.JsonMessageConverter;
+import org.springframework.cloud.function.json.JacksonMapper;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
@@ -42,4 +45,12 @@ public Consumer<UserMessage> logUserMessage() {
               userMessage.getUsername(), userMessage.getBody(), userMessage.getCreatedAt()));
     };
   }
+
+  // This is workaround for https://github.com/GoogleCloudPlatform/spring-cloud-gcp/issues/3139
+  // and should be removed once fix https://github.com/spring-cloud/spring-cloud-function/pull/1162
+  // is deployed
+  @Bean
+  public JsonMessageConverter customJsonMessageConverter(ObjectMapper objectMapper) {
+    return new JsonMessageConverter(new JacksonMapper(objectMapper));
+  }
 }