Avoid unnecessary re-partitioning due to map calls.

Fixing streams get unnecessarily flagged for re-partitioning
from map calls on KStream.

Resolves #412

Author: sobychacko

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

@@ -16,9 +16,7 @@
 
 package org.springframework.cloud.stream.binder.kafka.streams;
 
-import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.kstream.KStream;
-import org.apache.kafka.streams.kstream.KeyValueMapper;
 
 import org.springframework.cloud.stream.binding.StreamListenerParameterAdapter;
 import org.springframework.core.MethodParameter;
@@ -52,7 +50,7 @@ public KStream adapt(KStream<?, ?> bindingTarget, MethodParameter parameter) {
 		final Class<?> valueClass = (resolvableType.getGeneric(1).getRawClass() != null)
 				? (resolvableType.getGeneric(1).getRawClass()) : Object.class;
 		if (this.KafkaStreamsBindingInformationCatalogue.isUseNativeDecoding(bindingTarget)) {
-			return bindingTarget.map((KeyValueMapper) KeyValue::new);
+			return bindingTarget;
 		}
 		else {
 			return kafkaStreamsMessageConversionDelegate.deserializeOnInbound(valueClass, bindingTarget);

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

@@ -19,7 +19,6 @@
 import java.io.Closeable;
 import java.io.IOException;
 
-import org.apache.kafka.streams.KeyValue;
 import org.apache.kafka.streams.kstream.KStream;
 
 import org.springframework.cloud.stream.binding.StreamListenerResultAdapter;
@@ -38,7 +37,7 @@ public boolean supports(Class<?> resultType, Class<?> boundElement) {
 	@Override
 	@SuppressWarnings("unchecked")
 	public Closeable adapt(KStream streamListenerResult, KStreamBoundElementFactory.KStreamWrapper boundElement) {
-		boundElement.wrap(streamListenerResult.map(KeyValue::new));
+		boundElement.wrap(streamListenerResult);
 		return new NoOpCloseable();
 	}
 

spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToTableJoinIntegrationTests.java

@@ -129,31 +129,8 @@ public void testStreamToTable() throws Exception {
 				"--spring.cloud.stream.kafka.streams.bindings.input.consumer.applicationId=StreamToTableJoinIntegrationTests-abc",
 				"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString(),
 				"--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString())) {
-			// Input 1: Clicks per user (multiple records allowed per user).
-			List<KeyValue<String, Long>> userClicks = Arrays.asList(
-					new KeyValue<>("alice", 13L),
-					new KeyValue<>("bob", 4L),
-					new KeyValue<>("chao", 25L),
-					new KeyValue<>("bob", 19L),
-					new KeyValue<>("dave", 56L),
-					new KeyValue<>("eve", 78L),
-					new KeyValue<>("alice", 40L),
-					new KeyValue<>("fang", 99L)
-			);
-
-			Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
-			senderProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
-			senderProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, LongSerializer.class);
 
-			DefaultKafkaProducerFactory<String, Long> pf = new DefaultKafkaProducerFactory<>(senderProps);
-			KafkaTemplate<String, Long> template = new KafkaTemplate<>(pf, true);
-			template.setDefaultTopic("user-clicks-1");
-
-			for (KeyValue<String, Long> keyValue : userClicks) {
-				template.sendDefault(keyValue.key, keyValue.value);
-			}
-
-			// Input 2: Region per user (multiple records allowed per user).
+			// Input 1: Region per user (multiple records allowed per user).
 			List<KeyValue<String, String>> userRegions = Arrays.asList(
 					new KeyValue<>("alice", "asia"),   /* Alice lived in Asia originally... */
 					new KeyValue<>("bob", "americas"),
@@ -176,6 +153,30 @@ public void testStreamToTable() throws Exception {
 				template1.sendDefault(keyValue.key, keyValue.value);
 			}
 
+			// Input 2: Clicks per user (multiple records allowed per user).
+			List<KeyValue<String, Long>> userClicks = Arrays.asList(
+					new KeyValue<>("alice", 13L),
+					new KeyValue<>("bob", 4L),
+					new KeyValue<>("chao", 25L),
+					new KeyValue<>("bob", 19L),
+					new KeyValue<>("dave", 56L),
+					new KeyValue<>("eve", 78L),
+					new KeyValue<>("alice", 40L),
+					new KeyValue<>("fang", 99L)
+			);
+
+			Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
+			senderProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+			senderProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, LongSerializer.class);
+
+			DefaultKafkaProducerFactory<String, Long> pf = new DefaultKafkaProducerFactory<>(senderProps);
+			KafkaTemplate<String, Long> template = new KafkaTemplate<>(pf, true);
+			template.setDefaultTopic("user-clicks-1");
+
+			for (KeyValue<String, Long> keyValue : userClicks) {
+				template.sendDefault(keyValue.key, keyValue.value);
+			}
+
 			List<KeyValue<String, Long>> expectedClicksPerRegion = Arrays.asList(
 					new KeyValue<>("americas", 101L),
 					new KeyValue<>("europe", 109L),
@@ -267,19 +268,6 @@ public void testGlobalStartOffsetWithLatestAndIndividualBindingWthEarliest() thr
 				"--spring.cloud.stream.kafka.streams.binder.brokers=" + embeddedKafka.getBrokersAsString(),
 				"--spring.cloud.stream.kafka.streams.binder.zkNodes=" + embeddedKafka.getZookeeperConnectionString())) {
 			Thread.sleep(1000L);
-			// Input 1: Clicks per user (multiple records allowed per user).
-			List<KeyValue<String, Long>> userClicks1 = Arrays.asList(
-					new KeyValue<>("bob", 4L),
-					new KeyValue<>("chao", 25L),
-					new KeyValue<>("bob", 19L),
-					new KeyValue<>("dave", 56L),
-					new KeyValue<>("eve", 78L),
-					new KeyValue<>("fang", 99L)
-			);
-
-			for (KeyValue<String, Long> keyValue : userClicks1) {
-				template.sendDefault(keyValue.key, keyValue.value);
-			}
 
 			// Input 2: Region per user (multiple records allowed per user).
 			List<KeyValue<String, String>> userRegions = Arrays.asList(
@@ -304,6 +292,24 @@ public void testGlobalStartOffsetWithLatestAndIndividualBindingWthEarliest() thr
 				template1.sendDefault(keyValue.key, keyValue.value);
 			}
 
+
+
+			// Input 1: Clicks per user (multiple records allowed per user).
+			List<KeyValue<String, Long>> userClicks1 = Arrays.asList(
+					new KeyValue<>("bob", 4L),
+					new KeyValue<>("chao", 25L),
+					new KeyValue<>("bob", 19L),
+					new KeyValue<>("dave", 56L),
+					new KeyValue<>("eve", 78L),
+					new KeyValue<>("fang", 99L)
+			);
+
+			for (KeyValue<String, Long> keyValue : userClicks1) {
+				template.sendDefault(keyValue.key, keyValue.value);
+			}
+
+
+
 			List<KeyValue<String, Long>> expectedClicksPerRegion = Arrays.asList(
 					new KeyValue<>("americas", 101L),
 					new KeyValue<>("europe", 56L),