From 8a5eafa58f349345645c31dd2a465b0317c60709 Mon Sep 17 00:00:00 2001
From: Soby Chacko <chackos@vmware.com>
Date: Fri, 6 Jan 2023 11:52:35 -0500
Subject: [PATCH] KafkaNull check improvements

Related to f5e606dc55c25fa9c72e04e79d2e78bf477988d0

Resolves #983
---
 .../function/context/catalog/SimpleFunctionRegistry.java  | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java
index 1453a9102..2250e95dc 100644
--- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java
+++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java
@@ -879,7 +879,13 @@ public class SimpleFunctionRegistry implements FunctionRegistry {
 				logger.debug("Invoking function: " + this + "with input type: " + this.getInputType());
 			}
 
-			Object result = ((Function) this.target).apply(inputValue);
+			Object result;
+			if (inputValue != null && inputValue.getClass().getName().equals("org.springframework.kafka.support.KafkaNull")) {
+				result = ((Function) this.target).apply(null);
+			}
+			else {
+				result = ((Function) this.target).apply(inputValue);
+			}
 
 			if (result instanceof Publisher && functionInvocationHelper != null) {
 				result = this.postProcessFunction((Publisher) result, firstInputMessage);