Add support for handling special types

There are certain cases where conversion has to be skiped. One such case is KafkaNull type which has to be sent as is.

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/SimpleFunctionRegistry.java

@@ -787,6 +787,9 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
 						}
 					}
 					else if (!FunctionTypeUtils.isMessage(type)) {
+						if (this.payloadIsSpecialType(((Message<?>) value).getPayload())) {
+							return null;
+						}
 						convertedValue = ((Message<?>) convertedValue).getPayload();
 					}
 				}
@@ -836,7 +839,14 @@ public class SimpleFunctionRegistry implements FunctionRegistry, FunctionInspect
 			}
 			return rawType instanceof Class<?>
 				&& !(message.getPayload() instanceof Optional)
+				&& !this.payloadIsSpecialType(message.getPayload())
 				&& !(message.getPayload().getClass().isAssignableFrom(((Class<?>) rawType)));
 		}
+
+		private boolean payloadIsSpecialType(Object payload) {
+			return "org.springframework.kafka.support.KafkaNull".equals(payload.getClass().getName());
+		}
 	}
+
+
 }