From 1b80cfcf4c4db4bdbd5d5c7925e9d3f23746121b Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Sat, 20 Oct 2018 18:21:31 -0400 Subject: [PATCH] Avoid unnecessary re-partitioning due to map calls. Fixing streams get unnecessarily flagged for re-partitioning from map calls on KStream. Resolves #412 --- ...KStreamStreamListenerParameterAdapter.java | 4 +- .../KStreamStreamListenerResultAdapter.java | 3 +- .../StreamToTableJoinIntegrationTests.java | 80 ++++++++++--------- 3 files changed, 45 insertions(+), 42 deletions(-) diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamStreamListenerParameterAdapter.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamStreamListenerParameterAdapter.java index 3f4728f94..96567c0c5 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamStreamListenerParameterAdapter.java +++ b/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 @@ class KStreamStreamListenerParameterAdapter implements StreamListenerParameterAd 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); diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamStreamListenerResultAdapter.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamStreamListenerResultAdapter.java index f35b8e532..72f0ad005 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamStreamListenerResultAdapter.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KStreamStreamListenerResultAdapter.java @@ -19,7 +19,6 @@ package org.springframework.cloud.stream.binder.kafka.streams; 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 @@ class KStreamStreamListenerResultAdapter implements StreamListenerResultAdapter< @Override @SuppressWarnings("unchecked") public Closeable adapt(KStream streamListenerResult, KStreamBoundElementFactory.KStreamWrapper boundElement) { - boundElement.wrap(streamListenerResult.map(KeyValue::new)); + boundElement.wrap(streamListenerResult); return new NoOpCloseable(); } diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToTableJoinIntegrationTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToTableJoinIntegrationTests.java index 1f4593306..b594b31c6 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToTableJoinIntegrationTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/integration/StreamToTableJoinIntegrationTests.java @@ -129,7 +129,31 @@ public class StreamToTableJoinIntegrationTests { "--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). + + // Input 1: Region per user (multiple records allowed per user). + List> userRegions = Arrays.asList( + new KeyValue<>("alice", "asia"), /* Alice lived in Asia originally... */ + new KeyValue<>("bob", "americas"), + new KeyValue<>("chao", "asia"), + new KeyValue<>("dave", "europe"), + new KeyValue<>("alice", "europe"), /* ...but moved to Europe some time later. */ + new KeyValue<>("eve", "americas"), + new KeyValue<>("fang", "asia") + ); + + Map senderProps1 = KafkaTestUtils.producerProps(embeddedKafka); + senderProps1.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + senderProps1.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + + DefaultKafkaProducerFactory pf1 = new DefaultKafkaProducerFactory<>(senderProps1); + KafkaTemplate template1 = new KafkaTemplate<>(pf1, true); + template1.setDefaultTopic("user-regions-1"); + + for (KeyValue keyValue : userRegions) { + template1.sendDefault(keyValue.key, keyValue.value); + } + + // Input 2: Clicks per user (multiple records allowed per user). List> userClicks = Arrays.asList( new KeyValue<>("alice", 13L), new KeyValue<>("bob", 4L), @@ -153,29 +177,6 @@ public class StreamToTableJoinIntegrationTests { template.sendDefault(keyValue.key, keyValue.value); } - // Input 2: Region per user (multiple records allowed per user). - List> userRegions = Arrays.asList( - new KeyValue<>("alice", "asia"), /* Alice lived in Asia originally... */ - new KeyValue<>("bob", "americas"), - new KeyValue<>("chao", "asia"), - new KeyValue<>("dave", "europe"), - new KeyValue<>("alice", "europe"), /* ...but moved to Europe some time later. */ - new KeyValue<>("eve", "americas"), - new KeyValue<>("fang", "asia") - ); - - Map senderProps1 = KafkaTestUtils.producerProps(embeddedKafka); - senderProps1.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); - senderProps1.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class); - - DefaultKafkaProducerFactory pf1 = new DefaultKafkaProducerFactory<>(senderProps1); - KafkaTemplate template1 = new KafkaTemplate<>(pf1, true); - template1.setDefaultTopic("user-regions-1"); - - for (KeyValue keyValue : userRegions) { - template1.sendDefault(keyValue.key, keyValue.value); - } - List> expectedClicksPerRegion = Arrays.asList( new KeyValue<>("americas", 101L), new KeyValue<>("europe", 109L), @@ -267,19 +268,6 @@ public class StreamToTableJoinIntegrationTests { "--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> 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 keyValue : userClicks1) { - template.sendDefault(keyValue.key, keyValue.value); - } // Input 2: Region per user (multiple records allowed per user). List> userRegions = Arrays.asList( @@ -304,6 +292,24 @@ public class StreamToTableJoinIntegrationTests { template1.sendDefault(keyValue.key, keyValue.value); } + + + // Input 1: Clicks per user (multiple records allowed per user). + List> 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 keyValue : userClicks1) { + template.sendDefault(keyValue.key, keyValue.value); + } + + + List> expectedClicksPerRegion = Arrays.asList( new KeyValue<>("americas", 101L), new KeyValue<>("europe", 56L),