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);