Allow for binders without embedded headers to process 'originalContentType'
As part of the changes that led to support for native serialization/deserialization we've made the assumption that any payload that is not a `byte[]` is already fully processed and so will skip `deserializePayloadIfNecessary`. However, if the binder is based on endpoints that do their own deserialization and populate headers (as for example RabbitMQ and the upcoming JMS), the current changes skip the part where `originalContentType` is processed. We should be more lenient and allow `deserializePayloadIfNecessary` to execute if we have an `originalContentType` header.
diff --git a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
index 3aac7c84d..db24919d0 100644
--- a/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
+++ b/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
@@ -278,7 +278,8 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
 		@Override
 		@SuppressWarnings("unchecked")
 		protected Object handleRequestMessage(Message<?> requestMessage) {
-			if (!(requestMessage.getPayload() instanceof byte[])) {
+			if (!(requestMessage.getPayload() instanceof byte[])
+					&& !requestMessage.getHeaders().containsKey(BinderHeaders.BINDER_ORIGINAL_CONTENT_TYPE)) {
 				return requestMessage;
 			}
 			MessageValues messageValues;
['spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java']