Create topology before kafka streams start. #3172

chickenchickenlove · 2024-03-30T09:15:14Z

Introduction

Hi, Spring kafka Team.
This is chickenchickenlove, huge fan of you guys.
I saw this issue (#3020). i want to resolve this problem by myself!

Background

Allows to initialize/inject the Topology object without starting the Kafka streams. #3020
If Topology of kafka streams are initialized before StreamsBuilderFactoryBean#start(), it will be very helpful to Test code for spring kafka-streams. TopologyTestDriver is very efficient to test comparing to TestContainers.

To reviewer

I have tiny PR to solve Issue(Allows to initialize/inject the Topology object without starting the Kafka streams. #3020).
It might be a fairly simple fix, but it could look bad. IMHO, it's a better idea to register StreamsBuilder as a Bean in a different scope, however, it will introduce many changes to StreamsBuilderFactoryBean, AbstractBuilderFactoryBean as well. I have made modifications to minimize the changes.
When you have some free time, Please take a look 🙇‍♂️.

Changes

Point of initializing kafka-streams topology move to StreamsBuilderFactoryBean#isAutoStartup().

`@PostConstruct` or `StreamsBuilderFactoryBean#afterPropertiesSet()` Does not work well in this case. Why?

Because they are called just before bean is initialized. however, codes to make topology will be injected after StreamsBuilderFactoryBean#afterPropertiesSet() are called.

StreamBuilder is initialized in that time by this code. (StreamsBuilderFactoryBean#afterPropertiesSet() -> StreamsBuilderFactoryBean#createInstance() -> finally, StreamsBuilder are intialized!)
However, codes to make topology, are not injected StreamsBuilder in this time. you can refer to this topology code below.

@Component
public class MyComponent {

    private static final Serde<String> STRING_SERDE = Serdes.String();
    private final StreamsBuilder streamsBuilder;

    public MyComponent(StreamsBuilder streamsBuilder) {
        this.streamsBuilder = streamsBuilder;
        buildPipeline();
    }

    public void buildPipeline() {
        KStream<String, String> messageStream = streamsBuilder
                .stream("input-topic", Consumed.with(STRING_SERDE, STRING_SERDE));
        ...
        wordCounts.toStream().to("output-topic");
    }
}

MyComponent#buildPipeline()is executed afterStreamsBuilderFactoryBean#afterPropertiesSet()called. After this step,StreamsBuildercan return validtopology. Before this step, StreamsBuilderwill return invalidtopology` always.

~~### Why am i used to isAutoStart()?~~
~~- AFAIK, There is no hooking point for this. even, ContextRefreshedEvent will be called after SmartLifeCycle#isAutoStart()~~
- IMHO, this is best simple modification. In common use case, Spring will call SmartLifeCycle#isAutoStart() only once. I did send request to spring team whether if they can implement new hooking point such like it. (link)

Any Other workaround?

~~- IMHO, StreamsBuilder should be registered as bean other scope (not in StreamsBuilderFactoryBean scope). But it will involves a lot of changes to the StreamsBuilderFactoryBean.~~
~~- I was wrong. IMHO, there is no workaround, i guess 😢 .~~

Finally, i used SmartInitializingSingleton.afterSingletonsInstantiated() to resolve this issue.

sobychacko · 2024-04-01T14:47:25Z

@chickenchickenlove There are checkstyle issues and test failures in the build. Please take a look.

chickenchickenlove · 2024-04-01T15:05:56Z

Aye, i will check it. thanks for your comment 🙇‍♂️

chickenchickenlove · 2024-04-01T15:53:33Z

Hi, @sobychacko !
Thanks for your comments 🙇‍♂️
I fixed failed test and checkstyle!

Please take a look when you have some free time! 🙇‍♂️

sobychacko · 2024-04-01T18:09:57Z

spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBean.java

+		}
+		catch (Exception e) {
+			throw new RuntimeException(e);
+		}


It might be better to add this to the afterPropertiesSet() method instead of here. Doing it here adds more responsibilities to the isAutoStartup() method, which should technically only return the current autoStartup status. Also, we need to call the configureTopology(topology) method immediately after the topology creation so that some clients, such as the TopologyTestDriver, may use a fully configured topology.

Also, please add your name as an author of this class.

Looks like this comment is not addressed? https://github.com/spring-projects/spring-kafka/pull/3172/files#r1546662939

@sobychacko, I'm so sorry, i missed this comments.

It might be better to add this to the afterPropertiesSet() method instead of here. Doing it here adds more responsibilities to the isAutoStartup() method, which should technically only return the current autoStartup status. Also, we need to call the configureTopology(topology) method immediately after the topology creation so that some clients, such as the TopologyTestDriver, may use a fully configured topology.

IMHO, afterPropertiesSet() is not proper workaround.
streamBuilder#stream() seems to be called after StreamsBuilderFactoryBean#afterPropertiesSet() are executed.

When StreamsBuilderFactoryBean#afterPropertiesSet() is called, no stream declarations are made in the StreamBuilder yet. Therefore, executing StreamsBuilder.build() at the time of the afterPropertiesSet() call does not create any topology.

And, Unfortunately, it seems that there is no appropriate hooking point. (spring-projects/spring-framework#32554)

And i agree that isAutoStartup() has much responsibilities in this case.
If we move configureTopology(topology) to isAutoStartup(), much responsibilities as well.

However, as i said above, we cannot execute streamsBuilder#build() properly in scope of afterPropertiesSet().

Please refer to image above.

This is timeline of KafkaStreams with spring-kafka.
As you can see, In StreamsBuilderFactoryBean#afterPropertiesSet(), streamsBuilder is just about to be created and streamsBuilder does not have any stream DSL.

Thus streamsBuilder#build() is executed in StreamsBuilderFactoryBean#afterPropertiesSet(), there is no DSL to build on streamsBuilder.

Just commented on your issue in Spring Framework.
Let's see if SmartInitializingSingleton.afterSingletonsInstantiated() implementation help us somehow!

@artembilan Hi!
Thanks your comment. SmartInitializingSingleton.afterSingletonsInstantiated() works well.
I modified code to use SmartInitializingSingleton.afterSingletonsInstantiated() and test code as well.

Please Take a look. 🙇‍♂️

CC. @sobychacko

sobychacko · 2024-04-01T18:11:05Z

spring-kafka/src/test/java/org/springframework/kafka/config/StreamsBuilderFactoryBeanTests.java

@@ -92,20 +92,56 @@ public void testCleanupStreams() throws IOException {
 	}


Can you add your name as an author of this class?

Aye, i added my name classes below.

StreamsBuilderFactoryBean

StreamsBuilderFactoryBeanTests

StreamsBuilderFactoryLateConfigTests
Please take a look 🙇‍♂️

sobychacko · 2024-04-01T18:11:52Z

...fka/src/test/java/org/springframework/kafka/config/StreamsBuilderFactoryLateConfigTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2020 the original author or authors.
+ * Copyright 2018-2024 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.


Missing author tag.

sobychacko · 2024-04-02T00:10:15Z

spring-kafka/src/main/java/org/springframework/kafka/config/StreamsBuilderFactoryBean.java

+		}
+		catch (Exception e) {
+			throw new RuntimeException(e);
+		}


Looks like this comment is not addressed? https://github.com/spring-projects/spring-kafka/pull/3172/files#r1546662939

sobychacko · 2024-04-03T21:19:44Z

spring-kafka/src/test/java/org/springframework/kafka/config/StreamsBuilderFactoryBeanTests.java

+		};
+		streamsBuilderFactoryBean.afterPropertiesSet();
+		StreamsBuilder builder = streamsBuilderFactoryBean.getObject();
+		builder.stream(Pattern.compile("foo"));


Try not to use names like foo in the new code that we add.

sobychacko · 2024-04-03T21:20:56Z

@chickenchickenlove The new updates look good, but the PR build still has some checkstyle issues. See more details here: https://github.com/spring-projects/spring-kafka/actions/runs/8532654028/job/23411166278?pr=3172.
Once that is addressed, this is ready to be merged.

chickenchickenlove · 2024-04-03T22:03:37Z

@sobychacko
Thanks for your time 🙇‍♂️
I have resolved the lint error and changed from foo to test-topic.
Could you please take a look?

sobychacko · 2024-04-04T13:47:08Z

@chickenchickenlove Thanks for the PR. It is now merged upstream.

injae-kim · 2024-04-04T14:13:29Z

Nice work @chickenchickenlove !!

Create topology before kafka streams start.

193dbb7

chickenchickenlove mentioned this pull request Mar 30, 2024

Allows to initialize/inject the Topology object without starting the Kafka streams. #3020

Closed

Fix failed tests and fix check style

e65e6ff

sobychacko requested changes Apr 1, 2024

View reviewed changes

Add author

20ba75d

chickenchickenlove requested a review from sobychacko April 1, 2024 22:55

sobychacko requested changes Apr 2, 2024

View reviewed changes

chickenchickenlove requested a review from sobychacko April 2, 2024 01:42

Use SmartInitializingSingleton to build topology before start.

9fafa67

chickenchickenlove requested a review from artembilan April 3, 2024 03:11

sobychacko approved these changes Apr 3, 2024

View reviewed changes

sobychacko requested changes Apr 3, 2024

View reviewed changes

fix lint error, and do not use 'foo' in the new code

526392f

chickenchickenlove requested a review from sobychacko April 3, 2024 22:02

sobychacko merged commit f1e48f4 into spring-projects:main Apr 4, 2024
3 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Create topology before kafka streams start. #3172

Create topology before kafka streams start. #3172

chickenchickenlove commented Mar 30, 2024 •

edited

Loading

sobychacko commented Apr 1, 2024 •

edited

Loading

chickenchickenlove commented Apr 1, 2024

chickenchickenlove commented Apr 1, 2024

sobychacko Apr 1, 2024

sobychacko Apr 1, 2024

sobychacko Apr 2, 2024

chickenchickenlove Apr 2, 2024 •

edited

Loading

chickenchickenlove Apr 2, 2024 •

edited

Loading

artembilan Apr 2, 2024

chickenchickenlove Apr 3, 2024

sobychacko Apr 1, 2024

chickenchickenlove Apr 1, 2024

sobychacko Apr 1, 2024

sobychacko Apr 2, 2024

sobychacko Apr 3, 2024

sobychacko commented Apr 3, 2024

chickenchickenlove commented Apr 3, 2024 •

edited

Loading

sobychacko commented Apr 4, 2024

injae-kim commented Apr 4, 2024

		@@ -92,20 +92,56 @@ public void testCleanupStreams() throws IOException {
		}

Create topology before kafka streams start. #3172

Create topology before kafka streams start. #3172

Conversation

chickenchickenlove commented Mar 30, 2024 • edited Loading

Introduction

Background

To reviewer

Changes

@PostConstruct or StreamsBuilderFactoryBean#afterPropertiesSet() Does not work well in this case. Why?

Any Other workaround?

sobychacko commented Apr 1, 2024 • edited Loading

chickenchickenlove commented Apr 1, 2024

chickenchickenlove commented Apr 1, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

chickenchickenlove Apr 2, 2024 • edited Loading

Choose a reason for hiding this comment

chickenchickenlove Apr 2, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

sobychacko commented Apr 3, 2024

chickenchickenlove commented Apr 3, 2024 • edited Loading

sobychacko commented Apr 4, 2024

injae-kim commented Apr 4, 2024

chickenchickenlove commented Mar 30, 2024 •

edited

Loading

`@PostConstruct` or `StreamsBuilderFactoryBean#afterPropertiesSet()` Does not work well in this case. Why?

sobychacko commented Apr 1, 2024 •

edited

Loading

chickenchickenlove Apr 2, 2024 •

edited

Loading

chickenchickenlove Apr 2, 2024 •

edited

Loading

chickenchickenlove commented Apr 3, 2024 •

edited

Loading