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(); + } } /**