Changed occurrences of map calls on kafka streams to mapValues - Issue #357

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsMessageConversionDelegate.java

@@ -73,17 +73,17 @@ public KStream serializeOnOutbound(KStream<?,?> outboundBindTarget) {
 		String contentType = this.kstreamBindingInformationCatalogue.getContentType(outboundBindTarget);
 		MessageConverter messageConverter = compositeMessageConverterFactory.getMessageConverterForAllRegistered();
 
-		return outboundBindTarget.map((k, v) -> {
+		return outboundBindTarget.mapValues((v) -> {
 			Message<?> message = v instanceof Message<?> ? (Message<?>) v :
 					MessageBuilder.withPayload(v).build();
 			Map<String, Object> headers = new HashMap<>(message.getHeaders());
 			if (!StringUtils.isEmpty(contentType)) {
 				headers.put(MessageHeaders.CONTENT_TYPE, contentType);
 			}
 			MessageHeaders messageHeaders = new MessageHeaders(headers);
-			return new KeyValue<>(k,
+			return 
 					messageConverter.toMessage(message.getPayload(),
-							messageHeaders).getPayload());
+							messageHeaders).getPayload();
 		});
 	}
 
@@ -137,10 +137,10 @@ else if (o2 instanceof String || o2 instanceof byte[]) {
 		processErrorFromDeserialization(bindingTarget, branch[1]);
 
 		//first branch above is the branch where the messages are converted, let it go through further processing.
-		return branch[0].map((o, o2) -> {
-			KeyValue<Object, Object> objectObjectKeyValue = keyValueThreadLocal.get();
+		return branch[0].mapValues((o2) -> {
+			Object objectValue = keyValueThreadLocal.get().value;
 			keyValueThreadLocal.remove();
-			return objectObjectKeyValue;
+			return objectValue;
 		});
 	}
 

spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java

@@ -27,7 +27,6 @@
 import org.apache.kafka.common.serialization.Serde;
 import org.apache.kafka.common.utils.Bytes;
 import org.apache.kafka.streams.Consumed;
-import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.StreamsBuilder;
 import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.errors.DeserializationExceptionHandler;
@@ -300,18 +299,18 @@ private <K,V> KTable<K,V> materializedAs(StreamsBuilder streamsBuilder, String d
 		else {
 			LOG.info("Native decoding is disabled for " + inboundName + ". Inbound message conversion done by Spring Cloud Stream.");
 		}
-		stream = stream.map((key, value) -> {
-			KeyValue<Object, Object> keyValue;
+
+		stream = stream.mapValues(value -> {
+			Object returnValue;
 			String contentType = bindingProperties.getContentType();
 			if (!StringUtils.isEmpty(contentType) && !nativeDecoding) {
 				Message<?> message = MessageBuilder.withPayload(value)
 						.setHeader(MessageHeaders.CONTENT_TYPE, contentType).build();
-				keyValue = new KeyValue<>(key, message);
-			}
-			else {
-				keyValue = new KeyValue<>(key, value);
+						returnValue = message;
+			} else {
+				returnValue = value;
 			}
-			return keyValue;
+			return returnValue;
 		});
 		return stream;
 	}