From 34b0945d43d50190b5fbe03f5f168c92aa168cfe Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Mon, 11 Nov 2019 17:21:01 -0500 Subject: [PATCH] Changing the order of calling customizer In the Kafka Streams binder, StreamsBuilderFactoryBean customzier was being called prematurely before the object is created. Fixing this issue. Add a test to verify --- .../AbstractKafkaStreamsBinderProcessor.java | 20 +++++++------ ...kaStreamsBinderWordCountFunctionTests.java | 28 +++++++++++++++++-- 2 files changed, 38 insertions(+), 10 deletions(-) diff --git a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java index dc0b3931b..1e18a7879 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java +++ b/spring-cloud-stream-binder-kafka-streams/src/main/java/org/springframework/cloud/stream/binder/kafka/streams/AbstractKafkaStreamsBinderProcessor.java @@ -214,18 +214,16 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application KafkaStreamsConfiguration kafkaStreamsConfiguration = new KafkaStreamsConfiguration(streamConfigGlobalProperties); - StreamsBuilderFactoryBean streamsBuilder = this.cleanupConfig == null + StreamsBuilderFactoryBean streamsBuilderFactoryBean = this.cleanupConfig == null ? new StreamsBuilderFactoryBean(kafkaStreamsConfiguration) : new StreamsBuilderFactoryBean(kafkaStreamsConfiguration, this.cleanupConfig); - if (customizer != null) { - customizer.configure(streamsBuilder); - } - streamsBuilder.setAutoStartup(false); + + streamsBuilderFactoryBean.setAutoStartup(false); BeanDefinition streamsBuilderBeanDefinition = BeanDefinitionBuilder .genericBeanDefinition( - (Class) streamsBuilder.getClass(), - () -> streamsBuilder) + (Class) streamsBuilderFactoryBean.getClass(), + () -> streamsBuilderFactoryBean) .getRawBeanDefinition(); ((BeanDefinitionRegistry) beanFactory).registerBeanDefinition( "stream-builder-" + beanNamePostPrefix, streamsBuilderBeanDefinition); @@ -234,8 +232,14 @@ public abstract class AbstractKafkaStreamsBinderProcessor implements Application //Removing the application ID from global properties so that the next function won't re-use it and cause race conditions. streamConfigGlobalProperties.remove(StreamsConfig.APPLICATION_ID_CONFIG); - return applicationContext.getBean( + final StreamsBuilderFactoryBean streamsBuilderFactoryBeanFromContext = applicationContext.getBean( "&stream-builder-" + beanNamePostPrefix, StreamsBuilderFactoryBean.class); + //At this point, the StreamsBuilderFactoryBean is created. If the users call, getObject() + //in the customizer, that should grant access to the StreamsBuilder. + if (customizer != null) { + customizer.configure(streamsBuilderFactoryBean); + } + return streamsBuilderFactoryBeanFromContext; } protected Serde getValueSerde(String inboundName, KafkaStreamsConsumerProperties kafkaStreamsConsumerProperties, ResolvableType resolvableType) { diff --git a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java index 31b51561a..da4950926 100644 --- a/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java +++ b/spring-cloud-stream-binder-kafka-streams/src/test/java/org/springframework/cloud/stream/binder/kafka/streams/function/KafkaStreamsBinderWordCountFunctionTests.java @@ -19,6 +19,8 @@ package org.springframework.cloud.stream.binder.kafka.streams.function; import java.util.Arrays; import java.util.Date; import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import java.util.function.Function; import io.micrometer.core.instrument.MeterRegistry; @@ -27,9 +29,9 @@ import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.KeyValue; +import org.apache.kafka.streams.kstream.Grouped; import org.apache.kafka.streams.kstream.KStream; import org.apache.kafka.streams.kstream.Materialized; -import org.apache.kafka.streams.kstream.Serialized; import org.apache.kafka.streams.kstream.TimeWindows; import org.apache.kafka.streams.processor.StreamPartitioner; import org.junit.AfterClass; @@ -44,12 +46,14 @@ import org.springframework.boot.autoconfigure.EnableAutoConfiguration; import org.springframework.cloud.stream.binder.kafka.streams.InteractiveQueryService; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.Bean; +import org.springframework.kafka.config.StreamsBuilderFactoryBeanCustomizer; import org.springframework.kafka.core.DefaultKafkaConsumerFactory; import org.springframework.kafka.core.DefaultKafkaProducerFactory; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.test.EmbeddedKafkaBroker; import org.springframework.kafka.test.rule.EmbeddedKafkaRule; import org.springframework.kafka.test.utils.KafkaTestUtils; +import org.springframework.util.Assert; import static org.assertj.core.api.Assertions.assertThat; @@ -63,6 +67,8 @@ public class KafkaStreamsBinderWordCountFunctionTests { private static Consumer consumer; + private final static CountDownLatch LATCH = new CountDownLatch(1); + @BeforeClass public static void setUp() { Map consumerProps = KafkaTestUtils.consumerProps("group", "false", @@ -100,6 +106,7 @@ public class KafkaStreamsBinderWordCountFunctionTests { final MeterRegistry meterRegistry = context.getBean(MeterRegistry.class); Thread.sleep(100); assertThat(meterRegistry.get("stream.metrics.commit.total").gauge().value()).isEqualTo(1.0); + Assert.isTrue(LATCH.await(5, TimeUnit.SECONDS), "Failed to call customizers"); } } @@ -239,7 +246,7 @@ public class KafkaStreamsBinderWordCountFunctionTests { return input -> input .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+"))) .map((key, value) -> new KeyValue<>(value, value)) - .groupByKey(Serialized.with(Serdes.String(), Serdes.String())) + .groupByKey(Grouped.with(Serdes.String(), Serdes.String())) .windowedBy(TimeWindows.of(5000)) .count(Materialized.as("foo-WordCounts")) .toStream() @@ -247,6 +254,23 @@ public class KafkaStreamsBinderWordCountFunctionTests { new Date(key.window().start()), new Date(key.window().end())))); } + @Bean + public StreamsBuilderFactoryBeanCustomizer customizer() { + return fb -> { + try { + fb.setStateListener((newState, oldState) -> { + + }); + fb.getObject(); //make sure no exception is thrown at this call. + KafkaStreamsBinderWordCountFunctionTests.LATCH.countDown(); + + } + catch (Exception e) { + //Nothing to do - When the exception is thrown above, the latch won't be count down. + } + }; + } + @Bean public StreamPartitioner streamPartitioner() { return (t, k, v, n) -> k.equals("foo") ? 0 : 1;