diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java
index 1ff55d750..3d0f65235 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java
@@ -276,14 +276,17 @@ public class MessageConverterConfigurer implements MessageChannelAndSourceConfig
 		public Message<?> doPreSend(Message<?> message, MessageChannel channel) {
 			@SuppressWarnings("unchecked")
 			Map<String, Object> headersMap = (Map<String, Object>) ReflectionUtils.getField(MessageConverterConfigurer.this.headersField, message.getHeaders());
-
+			MimeType contentType = this.mimeType;
 			/*
 			 * NOTE: The below code for BINDER_ORIGINAL_CONTENT_TYPE is to support legacy message format established
 			 * in 1.x version of the framework and should/will no longer be supported in 3.x
 			 */
-			Object ct = message.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE);
-			MimeType contentType = ct instanceof String ? MimeType.valueOf((String)ct) : (ct == null ? this.mimeType : (MimeType)ct);
-			headersMap.remove(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE);
+			if (message.getHeaders().containsKey(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)) {
+				Object ct = message.getHeaders().get(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE);
+				contentType = ct instanceof String ? MimeType.valueOf((String)ct) : (ct == null ? this.mimeType : (MimeType)ct);
+				headersMap.put(MessageHeaders.CONTENT_TYPE, contentType);
+				headersMap.remove(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE);
+			}
 			// == end legacy note
 
 			if (!message.getHeaders().containsKey(MessageHeaders.CONTENT_TYPE)) {
diff --git a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java
index 9fb34cb80..8094745d1 100644
--- a/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java
+++ b/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/tck/ContentTypeTckTests.java
@@ -19,6 +19,7 @@ package org.springframework.cloud.stream.binder.tck;
 import java.nio.charset.StandardCharsets;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 
@@ -48,6 +49,7 @@ import org.springframework.messaging.MessageChannel;
 import org.springframework.messaging.MessageHeaders;
 import org.springframework.messaging.converter.AbstractMessageConverter;
 import org.springframework.messaging.converter.MessageConversionException;
+import org.springframework.messaging.handler.annotation.Payload;
 import org.springframework.messaging.handler.annotation.SendTo;
 import org.springframework.messaging.support.GenericMessage;
 import org.springframework.messaging.support.MessageBuilder;
@@ -71,6 +73,53 @@ import static org.junit.Assert.assertTrue;
  */
 public class ContentTypeTckTests {
 
+	@Test
+	public void stringToMapStreamListener() {
+		ApplicationContext context = new SpringApplicationBuilder(StringToMapStreamListener.class)
+				.web(WebApplicationType.NONE)
+				.run("--spring.jmx.enabled=false");
+		InputDestination source = context.getBean(InputDestination.class);
+		OutputDestination target = context.getBean(OutputDestination.class);
+		String jsonPayload = "{\"name\":\"oleg\"}";
+		source.send(new GenericMessage<byte[]>(jsonPayload.getBytes()));
+		Message<byte[]> outputMessage = target.receive();
+		assertEquals("oleg", new String(outputMessage.getPayload()));
+	}
+
+	@Test
+	public void stringToMapMessageStreamListener() {
+		ApplicationContext context = new SpringApplicationBuilder(StringToMapMessageStreamListener.class)
+				.web(WebApplicationType.NONE)
+				.run("--spring.jmx.enabled=false");
+		InputDestination source = context.getBean(InputDestination.class);
+		OutputDestination target = context.getBean(OutputDestination.class);
+		String jsonPayload = "{\"name\":\"oleg\"}";
+		source.send(new GenericMessage<byte[]>(jsonPayload.getBytes()));
+		Message<byte[]> outputMessage = target.receive();
+		assertEquals("oleg", new String(outputMessage.getPayload()));
+	}
+
+	@Test
+	// emulates 1.3 behavior
+	public void stringToMapMessageStreamListenerOriginalContentType() {
+		ApplicationContext context = new SpringApplicationBuilder(StringToMapMessageStreamListener.class)
+				.web(WebApplicationType.NONE)
+				.run("--spring.jmx.enabled=false");
+		InputDestination source = context.getBean(InputDestination.class);
+		OutputDestination target = context.getBean(OutputDestination.class);
+		String jsonPayload = "{\"name\":\"oleg\"}";
+
+		Message<byte[]> message = MessageBuilder.withPayload(jsonPayload.getBytes())
+			.setHeader(MessageHeaders.CONTENT_TYPE, "text/plain")
+			.setHeader("originalContentType", "application/json;charset=UTF-8")
+			.build();
+
+		source.send(message);
+		Message<byte[]> outputMessage = target.receive();
+		assertEquals("oleg", new String(outputMessage.getPayload()));
+	}
+
+
 	@Test
 	public void withInternalPipeline() {
 		ApplicationContext context = new SpringApplicationBuilder(InternalPipeLine.class)
@@ -505,6 +554,27 @@ public class ContentTypeTckTests {
 		}
 	}
 
+	@EnableBinding(Processor.class)
+	@Import(TestChannelBinderConfiguration.class)
+	public static class StringToMapStreamListener {
+		@StreamListener(Processor.INPUT)
+		@SendTo(Processor.OUTPUT)
+		public String echo(@Payload Map<?, ?> value)  {
+			return (String) value.get("name");
+		}
+	}
+
+	@EnableBinding(Processor.class)
+	@Import(TestChannelBinderConfiguration.class)
+	public static class StringToMapMessageStreamListener {
+		@StreamListener(Processor.INPUT)
+		@SendTo(Processor.OUTPUT)
+		public String echo(Message<Map<?, ?>> value)  {
+			assertTrue(value.getPayload() instanceof Map);
+			return (String) value.getPayload().get("name");
+		}
+	}
+
 	@EnableBinding(Processor.class)
 	@Import(TestChannelBinderConfiguration.class)
 	public static class PojoMessageToStringMessageStreamListener {