diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeUtils.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeUtils.java
index a06f23dc5..ffc8c8e15 100644
--- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeUtils.java
+++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/catalog/FunctionTypeUtils.java
@@ -44,6 +44,8 @@
 import java.util.function.ToLongFunction;
 
 import com.fasterxml.jackson.databind.JsonNode;
+import kotlin.jvm.functions.Function0;
+import kotlin.jvm.functions.Function1;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.reactivestreams.Publisher;
@@ -59,6 +61,7 @@
 import org.springframework.cloud.function.context.config.RoutingFunction;
 import org.springframework.context.support.GenericApplicationContext;
 import org.springframework.core.GenericTypeResolver;
+import org.springframework.core.KotlinDetector;
 import org.springframework.core.ResolvableType;
 import org.springframework.messaging.Message;
 import org.springframework.util.Assert;
@@ -203,25 +206,40 @@ public static Method discoverFunctionalMethod(Class<?> pojoFunctionClass) {
 	}
 
 	public static Type discoverFunctionTypeFromClass(Class<?> functionalClass) {
-		Type t = discoverFunctionTypeFromFunctionMethod(discoverFunctionalMethod(functionalClass));
-		if (t == null) {
-			ResolvableType resolvableFunctionType = ResolvableType.forClass(functionalClass);
-			List<ResolvableType> resolvedGenerics = new ArrayList<>();
-			if (resolvableFunctionType.hasGenerics()) {
-				for (ResolvableType generic : resolvableFunctionType.getGenerics()) {
-					if (generic.getType() instanceof TypeVariable) {
-						resolvedGenerics.add(ResolvableType.forClass(Object.class));
-					}
-					else {
-						resolvedGenerics.add(generic);
+		if (KotlinDetector.isKotlinPresent()) {
+			if (Function1.class.isAssignableFrom(functionalClass)) {
+				ResolvableType kotlinType = ResolvableType.forClass(functionalClass).as(Function1.class);
+				return GenericTypeResolver.resolveType(kotlinType.getType(), functionalClass);
+			}
+			else if (Function0.class.isAssignableFrom(functionalClass)) {
+				ResolvableType kotlinType = ResolvableType.forClass(functionalClass).as(Function0.class);
+				return GenericTypeResolver.resolveType(kotlinType.getType(), functionalClass);
+			}
+		}
+		Type typeToReturn = null;
+		if (Function.class.isAssignableFrom(functionalClass)) {
+			for (Type superInterface : functionalClass.getGenericInterfaces()) {
+				if (superInterface != null && !superInterface.equals(Object.class)) {
+					if (superInterface.toString().contains("KStream") && ResolvableType.forType(superInterface).getGeneric(1).isArray()) {
+						return null;
 					}
 				}
 			}
-			ResolvableType[] generics = resolvedGenerics.toArray(new ResolvableType[] {});
-
-			t = ResolvableType.forClassWithGenerics(functionalClass, generics).getType();
+			ResolvableType functionType = ResolvableType.forClass(functionalClass).as(Function.class);
+			typeToReturn = GenericTypeResolver.resolveType(functionType.getType(), functionalClass);
+		}
+		else if (Consumer.class.isAssignableFrom(functionalClass)) {
+			ResolvableType functionType = ResolvableType.forClass(functionalClass).as(Consumer.class);
+			typeToReturn = GenericTypeResolver.resolveType(functionType.getType(), functionalClass);
+		}
+		else if (Supplier.class.isAssignableFrom(functionalClass)) {
+			ResolvableType functionType = ResolvableType.forClass(functionalClass).as(Supplier.class);
+			typeToReturn = GenericTypeResolver.resolveType(functionType.getType(), functionalClass);
 		}
-		return t;
+//		else {
+//			typeToReturn = TypeResolver.reify(functionalClass);
+//		}
+		return typeToReturn;
 	}
 
 	/**
diff --git a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java
index 93ae1b732..e2037f58d 100644
--- a/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java
+++ b/spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/ContextFunctionCatalogAutoConfiguration.java
@@ -31,6 +31,8 @@
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
 import com.google.gson.Gson;
 import io.cloudevents.spring.messaging.CloudEventMessageConverter;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.BeanFactory;
@@ -91,6 +93,7 @@
 @AutoConfigureAfter(name = {"org.springframework.cloud.function.deployer.FunctionDeployerConfiguration"})
 public class ContextFunctionCatalogAutoConfiguration {
 
+	private static Log logger = LogFactory.getLog(ContextFunctionCatalogAutoConfiguration.class);
 	/**
 	 * The name of the property to specify desired JSON mapper. Available values are `jackson' and 'gson'.
 	 */
@@ -246,7 +249,44 @@ private JsonMapper jackson(ApplicationContext context) {
 			mapper.configure(SerializationFeature.FAIL_ON_EMPTY_BEANS, false);
 			mapper.configure(DeserializationFeature.FAIL_ON_TRAILING_TOKENS, true);
 			mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+			if (logger.isDebugEnabled()) {
+				logger.debug("ObjectMapper configuration: " + getConfigDetails(mapper));
+			}
 			return new JacksonMapper(mapper);
 		}
+
+		private static String getConfigDetails(ObjectMapper mapper) {
+			StringBuilder sb = new StringBuilder();
+
+			sb.append("Modules:\n");
+			if (mapper.getRegisteredModuleIds().isEmpty()) {
+				sb.append("\t").append("-none-").append("\n");
+			}
+			for (Object m : mapper.getRegisteredModuleIds()) {
+				sb.append("  ").append(m).append("\n");
+			}
+
+			sb.append("\nSerialization Features:\n");
+			for (SerializationFeature f : SerializationFeature.values()) {
+				sb.append("\t").append(f).append(" -> ")
+						.append(mapper.getSerializationConfig().hasSerializationFeatures(f.getMask()));
+				if (f.enabledByDefault()) {
+					sb.append(" (enabled by default)");
+				}
+				sb.append("\n");
+			}
+
+			sb.append("\nDeserialization Features:\n");
+			for (DeserializationFeature f : DeserializationFeature.values()) {
+				sb.append("\t").append(f).append(" -> ")
+						.append(mapper.getDeserializationConfig().hasDeserializationFeatures(f.getMask()));
+				if (f.enabledByDefault()) {
+					sb.append(" (enabled by default)");
+				}
+				sb.append("\n");
+			}
+
+			return sb.toString();
+		}
 	}
 }