[Enhancement request] SpringWolf to be compatible with Reactor Kafka

[KafkaProServerless]

Hello team,

Came into this project from this talk:
https://www.youtube.com/watch?v=DylvTW_ia4Y

If nothing else, this project seems to be very cool, very interesting.

We are using Kafka, and want to integrate with SpringWolf.

However, we are not using a regular Spring Kafka, but using reactor kafka https://github.com/reactor/reactor-kafka

Same question as in the talk around minute 33, can this project support reactor kafka?

Thank you

[github-actions]

Welcome to Springwolf. Thanks a lot for reporting your first issue. Please check out our contributors guide and feel free to join us on discord.

[sam0r040]

Hi @KafkaProServerless, thanks a lot for your interest in Springwolf and watching the video.
Springwolf heavily relies on Spring Framework and therefore we do not plan nor have the capacity to support other frameworks.

[KafkaProServerless]

Hello @sam0r040, thank you for the answer.
But reactor kafka is also relying on Spring Framework.
It is not like reactor kafka is something that has nothing to do with spring kafka, it is just the reactor version of it.

[sam0r040]

Hi @KafkaProServerless,
if you have Spring ecosystem available, then you can use Springwolf. We can imagine that the following could work for you:

Given the example consumer from ractor-kafka, you can use Springwolfs generic annotations like this:

    @AsyncListener(
            operation = @AsyncOperation(
                    channelName = "kafka-topic-name"
            )
    )
    private void receiveRecord(CountDownLatch latch, @Payload ReceiverRecord<Integer, PayloadDto> record) {
    }

    record PayloadDto(String myProperty) {
    }

We extracted the lambda into a method so that you can make use of the @Payload annotation so that Springwolf can automatically detect payload type.

Don not forget to configure extractable classes for ReceiverRecord to that the payload is extracted correctly from the second generic type:
springwolf.payload.extractable-classes.reactor.kafka.receiver.ReceiverRecord=1

See https://www.springwolf.dev/docs/configuration/documenting-consumers#asynclistener for more details.

[KafkaProServerless]

Let me try that!

[patpatpat123]

Hello!

I am glad to have found this post, as I am trying to accomplish the same thing.
Actually, I would like to help the community by adding an example, for the integration between reactor kafkaesque and Springwolf!

However, I am having difficulties making this work as well.
I prepared a minimal reproducible example, something which can be pasted to the example folder later.

This minimal example, with just 5 files, is what I am targeting to achieve.
https://github.com/patpatpat123/springwolfreactorkafka/tree/main

The first file is the pom:
https://github.com/patpatpat123/springwolfreactorkafka/blob/main/pom.xml#L33

As you can see in the pom, there is the dependency to reactor Kafka. Not spring kafka, not spring cloud stream, but just spring and reactor Kafka!

As mentioned, there won't be any kafkalistener annotation, or any spring cloud stream annotation, yet, it is a plain spring based project.

Also, in the pom, there are the dependencies springwolf-core springwolf-kafka springwolf-asyncapi to all at version 1.6.0

Are those dependencies enough? Too much?

Now, looking at the code: there are just three java files.
1 - Is the "main" file, the SpringBootApplication file, which has nothing interesting. -> SpringWolfReactorKafkaApplication.java
2 - is the configuration file, where you can see the "reactive configuration" KafkaReceiver.create(receiverOptions.subscription(Collections.singleton("the_topic"))) Here, it differs from the traditional Spring Kafka. By the way, please feel free to replace the kafka broker, and the topic name with your information, this projects should work "out of the box" -> SpringWolfReactorKafkaConsumerConfiguration
3 - this most important file, the actual reactive processing:
https://github.com/patpatpat123/springwolfreactorkafka/blob/main/src/main/java/org/example/SpringWolfReactorKafkaService.java#L21

    @Override
    @AsyncListener(operation = @AsyncOperation(channelName = "the_topic"))
    public void run(final String... args) {
        kafkaReceiver.receiveAutoAck().concatMap(message -> message).flatMap(this::processMessages).subscribe();
    }

You can see here the AsyncListener suggested above. The code is really just a reactive consumer which consumes from a topic reactively.

Finally, a properties files: https://github.com/patpatpat123/springwolfreactorkafka/blob/main/src/main/resources/application.properties#L1

With this example, I was topic to also see the UI in the Youtube presentation, as well as some of the features presented there.

Could you guys please help on pointing out what is missing?
This would be a nice addition to the example folders

Thank you!