1

We're using micronaut/kafka-streams. Using this framework in order to create a streams application you build something like this:

@Factory
public class FooTopologyConfig {
  @Singleton
  @Named
  public KStream<String, FooPojo> configureTopology {
    return builder.stream("foo-topic-in")
           .peek((k,v) -> System.out.println(String.format("key %s, value: %s", k,v))
           .to("foo-topic-out");
  }
}

This:

  • Receives a ConfiguredStreamBuilder (a very light wrapper around StreamsBuilder)
  • Build and return the stream (we're not actually sure how important returning the stream is, but that's a different question).

ConfiguredStreamBuilder::build() (which invokes the same on StreamsBuilder) is called later by the framework and the returned Topology is not made available for injection by Micronaut.

We want the Topology bean in order to log a description of the topology (via Topology::describe).

Is it safe to do the following?

  • Call ConfiguredStreamBuilder::build (and therefore StreamsBuilder::build) and use the returned instance of Topology to print a human readable description.
  • Allow the framework to call ConfiguredStreamBuilder::build for a second time later, and use the second instance of the returned topology to build the application.
Sparky
  • 2,694
  • 3
  • 21
  • 31

1 Answers1

1

There should be no problem calling build() multiple times. This is common in the internal code of Streams as well as in the tests.

To answer your other question. you only need the stream from builder.stream() operations if you want to expand on that branch of the topology later.

wcarlson
  • 216
  • 1
  • 9