diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionCatalogWrapper.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionCatalogWrapper.java
index ced49d860..f6f7b6701 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionCatalogWrapper.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionCatalogWrapper.java
@@ -21,8 +21,10 @@ import org.springframework.util.Assert;
 
 /**
  * @author David Turanski
+ *
+ * @since 2.1
  **/
-public class FunctionCatalogWrapper {
+class FunctionCatalogWrapper {
 
 	private final FunctionCatalog catalog;
 
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java
index f9c8b7135..044c9e658 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/FunctionInvoker.java
@@ -28,7 +28,9 @@ import org.springframework.cloud.function.context.FunctionType;
 import org.springframework.cloud.function.context.catalog.FunctionInspector;
 import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
 import org.springframework.messaging.Message;
+import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.converter.CompositeMessageConverter;
+import org.springframework.messaging.support.ErrorMessage;
 import org.springframework.util.Assert;
 
 /**
@@ -51,23 +53,41 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
 
 	private final CompositeMessageConverter messageConverter;
 
+	private final MessageChannel errorChannel;
+
 	FunctionInvoker(String functionName, FunctionCatalogWrapper functionCatalog, FunctionInspector functionInspector,
 			CompositeMessageConverterFactory compositeMessageConverterFactory) {
+		this(functionName, functionCatalog, functionInspector, compositeMessageConverterFactory, null);
+	}
+
+	FunctionInvoker(String functionName, FunctionCatalogWrapper functionCatalog, FunctionInspector functionInspector,
+			CompositeMessageConverterFactory compositeMessageConverterFactory, MessageChannel errorChannel) {
 		this.userFunction = functionCatalog.lookup(functionName);
 		Assert.isInstanceOf(Function.class, this.userFunction);
 		this.messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered();
 		FunctionType functionType = functionInspector.getRegistration(this.userFunction).getType();
 		this.inputClass = functionType.getInputType();
+		this.errorChannel = errorChannel;
 	}
 
 	@Override
 	public Flux<Message<O>> apply(Flux<Message<I>> input) {
-		AtomicReference<Message<I>> originalMessage = new AtomicReference<>();
+		AtomicReference<Message<I>> originalMessageRef = new AtomicReference<>();
 		return input
-				.doOnNext(originalMessage::set)        // to preserve the original message
+				.doOnNext(originalMessageRef::set)        // to preserve the original message
 				.map(this::resolveArgument)                // resolves argument type before invocation of user function
+				.onErrorContinue((exception, originalMessage) -> {
+					if (this.errorChannel != null) {
+						ErrorMessage em = new ErrorMessage(exception, (Message<?>) originalMessage);
+						logger.error(em);
+						this.errorChannel.send(em);
+					}
+					else {
+						exception.printStackTrace();
+					}
+				})
 				.transform(this.userFunction::apply)    // invoke user function
-				.map(resultMessage -> toMessage(resultMessage, originalMessage.get())); // create output message
+				.map(resultMessage -> toMessage(resultMessage, originalMessageRef.get())); // create output message
 	}
 
 	@SuppressWarnings("unchecked")
@@ -87,9 +107,12 @@ class FunctionInvoker<I, O> implements Function<Flux<Message<I>>, Flux<Message<O
 			logger.debug("Resolving input argument from message: " + message);
 		}
 
-		return (T) (shouldConvertFromMessage(message)
+		T argument = (T) (shouldConvertFromMessage(message)
 				? this.messageConverter.fromMessage(message, this.inputClass)
 				: message);
+
+		Assert.notNull(argument, "Failed to resolve argument type '" + this.inputClass + "' from message: " + message );
+		return argument;
 	}
 
 	private boolean shouldConvertFromMessage(Message<?> message) {
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java
index f05e1ec88..4b43ad51d 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/IntegrationFlowFunctionSupport.java
@@ -25,6 +25,7 @@ import org.reactivestreams.Publisher;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.cloud.function.context.catalog.FunctionInspector;
 import org.springframework.cloud.function.core.FluxSupplier;
 import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
@@ -52,6 +53,9 @@ public class IntegrationFlowFunctionSupport {
 
 	private final FunctionProperties functionProperties;
 
+	@Autowired
+	private MessageChannel errorChannel;
+
 	/**
 	 * @param functionCatalog
 	 * @param functionInspector
@@ -115,7 +119,7 @@ public class IntegrationFlowFunctionSupport {
 		if (StringUtils.hasText(this.functionProperties.getName())) {
 			FunctionInvoker<I,O> functionInvoker =
 					new FunctionInvoker<>(this.functionProperties.getName(), this.functionCatalog,
-							this.functionInspector, this.messageConverterFactory);
+							this.functionInspector, this.messageConverterFactory, this.errorChannel);
 
 			subscribeToInput(functionInvoker, flowBuilder.toReactivePublisher(), outputChannel::send);
 			return true;
diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java
index 919c9faae..fda4d81ca 100644
--- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java
+++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/SourceToFunctionsSupportTests.java
@@ -39,11 +39,13 @@ import org.springframework.cloud.stream.messaging.Source;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Import;
+import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.dsl.IntegrationFlow;
 import org.springframework.integration.dsl.IntegrationFlowBuilder;
 import org.springframework.integration.support.MessageBuilder;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.MessageHeaders;
+import org.springframework.messaging.PollableChannel;
 import org.springframework.util.Assert;
 import org.springframework.util.MimeTypeUtils;
 
@@ -52,6 +54,8 @@ import static org.hamcrest.CoreMatchers.allOf;
 import static org.hamcrest.CoreMatchers.endsWith;
 import static org.hamcrest.CoreMatchers.isA;
 import static org.hamcrest.Matchers.hasProperty;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 
 /**
  * @author Oleg Zhurakousky
@@ -87,6 +91,21 @@ public class SourceToFunctionsSupportTests {
 		}
 	}
 
+	@Test
+	public void testFailedInputTypeConversion() {
+		try (ConfigurableApplicationContext context =
+				new SpringApplicationBuilder(
+						TestChannelBinderConfiguration.getCompleteConfiguration(FunctionsConfigurationNoContentType.class))
+						.web(WebApplicationType.NONE)
+						.run("--spring.cloud.stream.function.name=toUpperCase|concatWithSelf",
+								"--spring.jmx.enabled=false")) {
+			PollableChannel errorChannel = context.getBean("errorChannel", PollableChannel.class);
+			OutputDestination target = context.getBean(OutputDestination.class);
+			assertNull(target.receive(1000));
+			assertNotNull(errorChannel.receive(1000));
+		}
+	}
+
 	@Test
 	public void testMessageSourceIsCreatedFromProvidedSupplier() {
 		try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
@@ -140,7 +159,7 @@ public class SourceToFunctionsSupportTests {
 			allOf(isA(BeanInstantiationException.class), hasProperty("cause", isA(IllegalArgumentException.class)),
 				hasProperty("message", endsWith("'doesNotExist' cannot be located."))));
 
-		ConfigurableApplicationContext context = new SpringApplicationBuilder(
+		new SpringApplicationBuilder(
 			TestChannelBinderConfiguration.getCompleteConfiguration(SupplierConfiguration.class)).web(
 			WebApplicationType.NONE)
 			.run("--spring.cloud.stream.function.name=doesNotExist", "--spring.jmx.enabled=false");
@@ -185,6 +204,27 @@ public class SourceToFunctionsSupportTests {
 
 	}
 
+	@EnableAutoConfiguration
+	@Import(ExistingMessageSourceConfigurationNoContentTypeSet.class)
+	public static class FunctionsConfigurationNoContentType {
+
+		@Bean
+		public PollableChannel errorChannel() {
+			return new QueueChannel(10);
+		}
+
+		@Bean
+		public Function<String, String> toUpperCase() {
+			return String::toUpperCase;
+		}
+
+		@Bean
+		public Function<String, String> concatWithSelf() {
+			return x -> x + ":" + x;
+		}
+
+	}
+
 	/**
 	 * This configuration essentially emulates our existing app-starters for Sources
 	 * and essentially demonstrates how a function(s) could be applied to an existing
@@ -213,6 +253,28 @@ public class SourceToFunctionsSupportTests {
 
 	}
 
+	@EnableBinding(Source.class)
+	public static class ExistingMessageSourceConfigurationNoContentTypeSet {
+
+		@Autowired
+		private Source source;
+
+		@Bean
+		public IntegrationFlow messageSourceFlow(IntegrationFlowFunctionSupport functionSupport) {
+			Supplier<Message<String>> messageSource = () -> MessageBuilder.withPayload("hello function")
+					.setHeader(MessageHeaders.CONTENT_TYPE, "application/octet-stream").build();
+
+			IntegrationFlowBuilder flowBuilder = functionSupport.integrationFlowFromProvidedSupplier(messageSource);
+
+			if (!functionSupport.andThenFunction(flowBuilder, this.source.output())) {
+				flowBuilder = flowBuilder.channel(this.source.output());
+			}
+
+			return flowBuilder.get();
+		}
+
+	}
+
 	@EnableBinding(Source.class)
 	public static class ProvidedMessageSourceConfiguration {