SCST-GH-2355 Add support for batch processing of collections with different types

Specifically KafkaNull. But this commit effectively paves a path for any type

Author: olegz

spring-cloud-function-context/src/main/java/org/springframework/cloud/function/context/config/SmartCompositeMessageConverter.java

@@ -92,26 +92,30 @@ public Object fromMessage(Message<?> message, Class<?> targetClass, @Nullable Ob
 					isConverted = true;
 				}
 				for (Iterator<MessageConverter> iterator = getConverters().iterator(); iterator.hasNext() && !isConverted;) {
-					Message<?> m  = MessageBuilder.withPayload(item).copyHeaders(message.getHeaders()).build(); // TODO Message creating may be expensive
 					MessageConverter converter = (MessageConverter) iterator.next();
-					Object conversionResult = (converter instanceof SmartMessageConverter & genericItemRawType != genericItemType ?
-							((SmartMessageConverter) converter).fromMessage(m, genericItemRawType, genericItemType) :
-							converter.fromMessage(m, genericItemRawType));
-					if (conversionResult != null) {
-						resultList.add(conversionResult);
-						isConverted = true;
+					if (!converter.getClass().getName().endsWith("ApplicationJsonMessageMarshallingConverter")) { // TODO Stream stuff, needs to be removed
+						Message<?> m  = MessageBuilder.withPayload(item).copyHeaders(message.getHeaders()).build(); // TODO Message creating may be expensive
+						Object conversionResult = (converter instanceof SmartMessageConverter & genericItemRawType != genericItemType ?
+								((SmartMessageConverter) converter).fromMessage(m, genericItemRawType, genericItemType) :
+								converter.fromMessage(m, genericItemRawType));
+						if (conversionResult != null) {
+							resultList.add(conversionResult);
+							isConverted = true;
+						}
 					}
 				}
 			}
 			result = resultList;
 		}
 		else {
 			for (MessageConverter converter : getConverters()) {
-				result = (converter instanceof SmartMessageConverter ?
-						((SmartMessageConverter) converter).fromMessage(message, targetClass, conversionHint) :
-						converter.fromMessage(message, targetClass));
-				if (result != null) {
-					return result;
+				if (!converter.getClass().getName().endsWith("ApplicationJsonMessageMarshallingConverter")) {// TODO Stream stuff, needs to be removed
+					result = (converter instanceof SmartMessageConverter ?
+							((SmartMessageConverter) converter).fromMessage(message, targetClass, conversionHint) :
+							converter.fromMessage(message, targetClass));
+					if (result != null) {
+						return result;
+					}
 				}
 			}
 		}