From c43e45c7ada789125d66fefa072ab362d3552cda Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Fri, 19 Oct 2018 13:29:15 -0400 Subject: [PATCH] Addressing startOffset in Kafka Streams binder (#473) * Addressing startOffset in Kafka Streams binder Fixing the issue where auto.offset.reset is not honored through startOffset provided by the individual consumer binding. Adding integration test to override auto.offset.reset globally at the binder level to latest and then set startOffset on individual binding to earliest. Resolves #467 * Addressing PR review comments * Addressing PR review comments --- .../src/main/asciidoc/kafka-streams.adoc | 9 + ...StreamListenerSetupMethodOrchestrator.java | 70 +++++-- .../StreamToTableJoinIntegrationTests.java | 178 +++++++++++++++--- 3 files changed, 215 insertions(+), 42 deletions(-) diff --git a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/kafka-streams.adoc b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/kafka-streams.adoc index 1ad8c9308..b9c2fd588 100644 --- a/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/kafka-streams.adoc +++ b/spring-cloud-stream-binder-kafka-docs/src/main/asciidoc/kafka-streams.adoc @@ -163,6 +163,15 @@ dlqName:: DLQ topic name. + Default: `none`. +startOffset:: + Offset to start from if there is no committed offset to consume from. + This is mostly used when the consumer is consuming from a topic for the first time. Kafka Streams uses `earliest` as the default strategy and + the binder uses the same default. This can be overridden to `latest` using this property. ++ +Default: `earliest`. + +Note: Using `resetOffsets` on the consumer does not have any effect on Kafka Streams binder. +Unlike the message channel based binder, Kafka Streams binder does not seek to beginning or end on demand. === TimeWindow properties: diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java index e4cc229f7..beded2e45 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/KafkaStreamsStreamListenerSetupMethodOrchestrator.java @@ -29,6 +29,7 @@ import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.common.utils.Bytes; import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.StreamsConfig; +import org.apache.kafka.streams.Topology; import org.apache.kafka.streams.kstream.Consumed; import org.apache.kafka.streams.kstream.GlobalKTable; import org.apache.kafka.streams.kstream.KStream; @@ -47,6 +48,7 @@ import org.springframework.beans.factory.support.BeanDefinitionRegistry; import org.springframework.cloud.stream.annotation.Input; import org.springframework.cloud.stream.annotation.StreamListener; import org.springframework.cloud.stream.binder.ConsumerProperties; +import org.springframework.cloud.stream.binder.kafka.properties.KafkaConsumerProperties; import org.springframework.cloud.stream.binder.kafka.streams.annotations.KafkaStreamsStateStore; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsConsumerProperties; import org.springframework.cloud.stream.binder.kafka.streams.properties.KafkaStreamsExtendedBindingProperties; @@ -241,8 +243,26 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene KafkaStreamsStateStoreProperties spec = buildStateStoreSpec(method); Serde keySerde = this.keyValueSerdeResolver.getInboundKeySerde(extendedConsumerProperties); Serde valueSerde = this.keyValueSerdeResolver.getInboundValueSerde(bindingProperties.getConsumer(), extendedConsumerProperties); + + final KafkaConsumerProperties.StartOffset startOffset = extendedConsumerProperties.getStartOffset(); + Topology.AutoOffsetReset autoOffsetReset = null; + if (startOffset != null) { + switch (startOffset) { + case earliest : autoOffsetReset = Topology.AutoOffsetReset.EARLIEST; + break; + case latest : autoOffsetReset = Topology.AutoOffsetReset.LATEST; + break; + default: break; + } + } + if (extendedConsumerProperties.isResetOffsets()) { + LOG.warn("Detected resetOffsets configured on binding " + inboundName + ". " + + "Setting resetOffsets in Kafka Streams binder does not have any effect."); + } + if (parameterType.isAssignableFrom(KStream.class)) { - KStream stream = getkStream(inboundName, spec, bindingProperties, streamsBuilder, keySerde, valueSerde); + KStream stream = getkStream(inboundName, spec, bindingProperties, + streamsBuilder, keySerde, valueSerde, autoOffsetReset); KStreamBoundElementFactory.KStreamWrapper kStreamWrapper = (KStreamBoundElementFactory.KStreamWrapper) targetBean; //wrap the proxy created during the initial target type binding with real object (KStream) kStreamWrapper.wrap((KStream) stream); @@ -262,10 +282,8 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene else if (parameterType.isAssignableFrom(KTable.class)) { String materializedAs = extendedConsumerProperties.getMaterializedAs(); String bindingDestination = bindingServiceProperties.getBindingDestination(inboundName); - KTable table = materializedAs != null ? - materializedAs(streamsBuilder, bindingDestination, materializedAs, keySerde, valueSerde ) : - streamsBuilder.table(bindingDestination, - Consumed.with(keySerde, valueSerde)); + KTable table = getKTable(streamsBuilder, keySerde, valueSerde, materializedAs, + bindingDestination, autoOffsetReset); KTableBoundElementFactory.KTableWrapper kTableWrapper = (KTableBoundElementFactory.KTableWrapper) targetBean; //wrap the proxy created during the initial target type binding with real object (KTable) kTableWrapper.wrap((KTable) table); @@ -275,10 +293,8 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene else if (parameterType.isAssignableFrom(GlobalKTable.class)) { String materializedAs = extendedConsumerProperties.getMaterializedAs(); String bindingDestination = bindingServiceProperties.getBindingDestination(inboundName); - GlobalKTable table = materializedAs != null ? - materializedAsGlobalKTable(streamsBuilder, bindingDestination, materializedAs, keySerde, valueSerde ) : - streamsBuilder.globalTable(bindingDestination, - Consumed.with(keySerde, valueSerde)); + GlobalKTable table = getGlobalKTable(streamsBuilder, keySerde, valueSerde, materializedAs, + bindingDestination, autoOffsetReset); GlobalKTableBoundElementFactory.GlobalKTableWrapper globalKTableWrapper = (GlobalKTableBoundElementFactory.GlobalKTableWrapper) targetBean; //wrap the proxy created during the initial target type binding with real object (KTable) globalKTableWrapper.wrap((GlobalKTable) table); @@ -297,13 +313,33 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene return arguments; } - private KTable materializedAs(StreamsBuilder streamsBuilder, String destination, String storeName, Serde k, Serde v) { + private GlobalKTable getGlobalKTable(StreamsBuilder streamsBuilder, Serde keySerde, Serde valueSerde, String materializedAs, + String bindingDestination, Topology.AutoOffsetReset autoOffsetReset) { + return materializedAs != null ? + materializedAsGlobalKTable(streamsBuilder, bindingDestination, materializedAs, keySerde, valueSerde, autoOffsetReset) : + streamsBuilder.globalTable(bindingDestination, + Consumed.with(keySerde, valueSerde).withOffsetResetPolicy(autoOffsetReset)); + } + + private KTable getKTable(StreamsBuilder streamsBuilder, Serde keySerde, Serde valueSerde, String materializedAs, + String bindingDestination, Topology.AutoOffsetReset autoOffsetReset) { + return materializedAs != null ? + materializedAs(streamsBuilder, bindingDestination, materializedAs, keySerde, valueSerde, autoOffsetReset ) : + streamsBuilder.table(bindingDestination, + Consumed.with(keySerde, valueSerde).withOffsetResetPolicy(autoOffsetReset)); + } + + private KTable materializedAs(StreamsBuilder streamsBuilder, String destination, String storeName, Serde k, Serde v, + Topology.AutoOffsetReset autoOffsetReset) { return streamsBuilder.table(bindingServiceProperties.getBindingDestination(destination), + Consumed.with(k,v).withOffsetResetPolicy(autoOffsetReset), getMaterialized(storeName, k, v)); } - private GlobalKTable materializedAsGlobalKTable(StreamsBuilder streamsBuilder, String destination, String storeName, Serde k, Serde v) { + private GlobalKTable materializedAsGlobalKTable(StreamsBuilder streamsBuilder, String destination, String storeName, Serde k, Serde v, + Topology.AutoOffsetReset autoOffsetReset) { return streamsBuilder.globalTable(bindingServiceProperties.getBindingDestination(destination), + Consumed.with(k,v).withOffsetResetPolicy(autoOffsetReset), getMaterialized(storeName, k, v)); } @@ -349,8 +385,9 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene } private KStream getkStream(String inboundName, KafkaStreamsStateStoreProperties storeSpec, - BindingProperties bindingProperties, StreamsBuilder streamsBuilder, - Serde keySerde, Serde valueSerde) { + BindingProperties bindingProperties, + StreamsBuilder streamsBuilder, + Serde keySerde, Serde valueSerde, Topology.AutoOffsetReset autoOffsetReset) { if (storeSpec != null) { StoreBuilder storeBuilder = buildStateStore(storeSpec); streamsBuilder.addStateStore(storeBuilder); @@ -360,8 +397,11 @@ class KafkaStreamsStreamListenerSetupMethodOrchestrator implements StreamListene } String[] bindingTargets = StringUtils .commaDelimitedListToStringArray(bindingServiceProperties.getBindingDestination(inboundName)); - KStream stream = streamsBuilder.stream(Arrays.asList(bindingTargets), - Consumed.with(keySerde, valueSerde)); + + KStream stream = + streamsBuilder.stream(Arrays.asList(bindingTargets), + Consumed.with(keySerde, valueSerde) + .withOffsetResetPolicy(autoOffsetReset)); final boolean nativeDecoding = bindingServiceProperties.getConsumerProperties(inboundName).isUseNativeDecoding(); if (nativeDecoding){ LOG.info("Native decoding is enabled for " + inboundName + ". Inbound deserialization done at the broker."); 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 bb9cea6f3..1f4593306 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 @@ -36,8 +36,6 @@ import org.apache.kafka.streams.kstream.Joined; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.KTable; import org.apache.kafka.streams.kstream.Serialized; -import org.junit.AfterClass; -import org.junit.BeforeClass; import org.junit.ClassRule; import org.junit.Test; @@ -67,28 +65,10 @@ import static org.assertj.core.api.Assertions.assertThat; public class StreamToTableJoinIntegrationTests { @ClassRule - public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, "output-topic"); + public static EmbeddedKafkaRule embeddedKafkaRule = new EmbeddedKafkaRule(1, true, "output-topic-1", "output-topic-2"); private static EmbeddedKafkaBroker embeddedKafka = embeddedKafkaRule.getEmbeddedKafka(); - private static Consumer consumer; - - @BeforeClass - public static void setUp() throws Exception { - Map consumerProps = KafkaTestUtils.consumerProps("group", "false", embeddedKafka); - consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); - consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); - consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class); - DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); - consumer = cf.createConsumer(); - embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "output-topic"); - } - - @AfterClass - public static void tearDown() { - consumer.close(); - } - @EnableBinding(KafkaStreamsProcessorX.class) @EnableAutoConfiguration @EnableConfigurationProperties(KafkaStreamsApplicationSupportProperties.class) @@ -116,15 +96,24 @@ public class StreamToTableJoinIntegrationTests { } @Test - public void testStreamToTable() { + public void testStreamToTable() throws Exception { SpringApplication app = new SpringApplication(CountClicksPerRegionApplication.class); app.setWebApplicationType(WebApplicationType.NONE); + Consumer consumer; + Map consumerProps = KafkaTestUtils.consumerProps("group-1", "false", embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + consumer = cf.createConsumer(); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "output-topic-1"); + try (ConfigurableApplicationContext ignored = app.run("--server.port=0", "--spring.jmx.enabled=false", - "--spring.cloud.stream.bindings.input.destination=user-clicks", - "--spring.cloud.stream.bindings.input-x.destination=user-regions", - "--spring.cloud.stream.bindings.output.destination=output-topic", + "--spring.cloud.stream.bindings.input.destination=user-clicks-1", + "--spring.cloud.stream.bindings.input-x.destination=user-regions-1", + "--spring.cloud.stream.bindings.output.destination=output-topic-1", "--spring.cloud.stream.bindings.input.consumer.useNativeDecoding=true", "--spring.cloud.stream.bindings.input-x.consumer.useNativeDecoding=true", "--spring.cloud.stream.bindings.output.producer.useNativeEncoding=true", @@ -158,7 +147,7 @@ public class StreamToTableJoinIntegrationTests { DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf, true); - template.setDefaultTopic("user-clicks"); + template.setDefaultTopic("user-clicks-1"); for (KeyValue keyValue : userClicks) { template.sendDefault(keyValue.key, keyValue.value); @@ -181,7 +170,7 @@ public class StreamToTableJoinIntegrationTests { DefaultKafkaProducerFactory pf1 = new DefaultKafkaProducerFactory<>(senderProps1); KafkaTemplate template1 = new KafkaTemplate<>(pf1, true); - template1.setDefaultTopic("user-regions"); + template1.setDefaultTopic("user-regions-1"); for (KeyValue keyValue : userRegions) { template1.sendDefault(keyValue.key, keyValue.value); @@ -208,6 +197,141 @@ public class StreamToTableJoinIntegrationTests { assertThat(count == expectedClicksPerRegion.size()).isTrue(); assertThat(actualClicksPerRegion).hasSameElementsAs(expectedClicksPerRegion); } + finally { + consumer.close(); + } + } + + @Test + public void testGlobalStartOffsetWithLatestAndIndividualBindingWthEarliest() throws Exception { + SpringApplication app = new SpringApplication(CountClicksPerRegionApplication.class); + app.setWebApplicationType(WebApplicationType.NONE); + + Consumer consumer; + Map consumerProps = KafkaTestUtils.consumerProps("group-2", "false", embeddedKafka); + consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); + consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); + consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class); + DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(consumerProps); + consumer = cf.createConsumer(); + embeddedKafka.consumeFromAnEmbeddedTopic(consumer, "output-topic-2"); + + // Produce data first to the input topic to test the startOffset setting on the + // binding (which is set to earliest below). + // Input 1: Clicks per user (multiple records allowed per user). + List> userClicks = Arrays.asList( + new KeyValue<>("alice", 100L), + new KeyValue<>("alice", 100L), + new KeyValue<>("alice", 100L), + new KeyValue<>("alice", 100L), + new KeyValue<>("alice", 100L), + new KeyValue<>("alice", 100L), + new KeyValue<>("alice", 100L), + new KeyValue<>("alice", 100L), + new KeyValue<>("alice", 100L), + new KeyValue<>("alice", 100L) + ); + + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + senderProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class); + senderProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, LongSerializer.class); + + DefaultKafkaProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic("user-clicks-2"); + + for (KeyValue keyValue : userClicks) { + template.sendDefault(keyValue.key, keyValue.value); + } + //Thread.sleep(10000L); + try (ConfigurableApplicationContext ignored = app.run("--server.port=0", + "--spring.jmx.enabled=false", + "--spring.cloud.stream.bindings.input.destination=user-clicks-2", + "--spring.cloud.stream.bindings.input-x.destination=user-regions-2", + "--spring.cloud.stream.bindings.output.destination=output-topic-2", + "--spring.cloud.stream.bindings.input.consumer.useNativeDecoding=true", + "--spring.cloud.stream.bindings.input-x.consumer.useNativeDecoding=true", + "--spring.cloud.stream.bindings.output.producer.useNativeEncoding=true", + "--spring.cloud.stream.kafka.streams.binder.configuration.auto.offset.reset=latest", + "--spring.cloud.stream.kafka.streams.bindings.input.consumer.startOffset=earliest", + "--spring.cloud.stream.kafka.streams.bindings.input.consumer.keySerde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.bindings.input.consumer.valueSerde=org.apache.kafka.common.serialization.Serdes$LongSerde", + "--spring.cloud.stream.kafka.streams.bindings.inputX.consumer.keySerde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.bindings.inputX.consumer.valueSerde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.bindings.output.producer.keySerde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.bindings.output.producer.valueSerde=org.apache.kafka.common.serialization.Serdes$LongSerde", + "--spring.cloud.stream.kafka.streams.binder.configuration.default.key.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.binder.configuration.default.value.serde=org.apache.kafka.common.serialization.Serdes$StringSerde", + "--spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=10000", + "--spring.cloud.stream.kafka.streams.bindings.input.consumer.application-id=helloxyz-foobar", + "--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( + 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-2"); + + for (KeyValue keyValue : userRegions) { + template1.sendDefault(keyValue.key, keyValue.value); + } + + List> expectedClicksPerRegion = Arrays.asList( + new KeyValue<>("americas", 101L), + new KeyValue<>("europe", 56L), + new KeyValue<>("asia", 124L), + //1000 alice entries which were there in the topic before the consumer started. + //Since we set the startOffset to earliest for the topic, it will read them, + //but the join fails to associate with a valid region, thus UNKNOWN. + new KeyValue<>("UNKNOWN", 1000L) + ); + + //Verify that we receive the expected data + int count = 0; + long start = System.currentTimeMillis(); + List> actualClicksPerRegion = new ArrayList<>(); + do { + ConsumerRecords records = KafkaTestUtils.getRecords(consumer); + count = count + records.count(); + for (ConsumerRecord record : records) { + actualClicksPerRegion.add(new KeyValue<>(record.key(), record.value())); + } + } while (count < expectedClicksPerRegion.size() && (System.currentTimeMillis() - start) < 30000); + + assertThat(count).isEqualTo(expectedClicksPerRegion.size()); + assertThat(actualClicksPerRegion).hasSameElementsAs(expectedClicksPerRegion); + } + finally { + consumer.close(); + } } /**