How to start a Kafka-Streams Pipeline by Topology using Quarkus

I am following the Quarkus Kafka-Streams tutorial and can't quite understand how a pipeline can be started.

In the tutorial the org.apache.kafka.streams.StreamsBuilder is used to build a org.apache.kafka.streams.Topology that describes the pipeline. The method that builds the Topology is annotated with @Produces. In this cheat sheet it's described that this is sufficient to run a Kafka-Streams pipeline. In the tutorial an http-endpoint is exposed additionally. That's not required in the service I'm currently implementing. Also in the example the provider method is never called explicitly. When I start the application without the endpoint the pipeline is not started.

In this tutorial the pipeline is instantiated explicitly with a topology. But here properties have to be set manually and configuration is not taken from the quarkus.kafka-streams.<something> properties.

The question is: How can I use Topology builder from the first tutorial to start the pipeline described by it? Optimal case would be that the configuration from quarkus.kafka-streams.<something> is automatically applied.

Using:

• Java OpenJDK 11.0.8
• Quarkus Version: 1.8.0.Final

Figured out the problem: No guarantee for complete correctness. Just want to share what solved my problem.

Most important is to use the right @Produces. javax.enterprise.inject.Produces is necessary to be used for the Topology producing method. javax.ws.rs.Produces can be used additionally to define the MediaType of the output, but is not mandatory:

    @javax.ws.rs.Produces( MediaType.TEXT_PLAIN )
    @Produces
    @AlternativePriority( 1 )
    public Topology buildTopology() {
        ...
    }

An instance of Kafka Streams is automatically built by the framework on start-up. All you need to run your pipeline is the following:

@ApplicationScoped
public class YourApplication {
    private final KafkaStreams streams;

    public NasDistributorApplication( final KafkaStreams streams ) {
        this.streams = streams;
    }

    public void onStart( @Observes final StartupEvent startupEvent ) {
        streams.start();
    }

    public void onStop( @Observes final ShutdownEvent shutdownEvent ) {
        streams.close();
    }

}

Annotating the Topology producing method by @AlternativePriority( 1 ) might be necessary to tell quarkus to use this method to build the Topology. I don't know the internals of the framework but what I suspect that a default Topology is used and @AlternativePriority gives a higher priority to the custom method than the default Topology