Supporting reactor-kafka

[ghahramani]

Hi,

After some discussion in the Gitter spring cloud stream channel, I realized to have a sequential pause/resume situation in reactive Kafka we need to use a back pressure mechanism that is not supported by web flux and Stream right now as we do not have any binder for reactor-kafka yet. I would like to raise this feature request for supporting reactor-kafka in spring cloud stream so we can easily use its functionalities.

[garyrussell]

@sabbyanandan @olegz This may not be trivial, but I think if we can pull it off, it should become the only way to support reactive applications in SCSt.

Reactor apps on top of non-reactive binders (especially Kafka) is a fool's errand in my opinion.

There is also a reactor-rabbitmq project.

Perhaps we should spike this?

[olegz]

@garyrussell I completely agree and believe made it clear internally and externally. The same I believe would need to be made for Rabbit https://github.com/reactor/reactor-rabbitmq. This would also allow us to partition the binder API on reactive and imperative where reactive can still support the imperative model but not the other way around as it doesn't make any sense (as today)

[sabbyanandan]

@ghahramani: When you get a chance, can you describe your use-case, and why the backpressure and being fully reactive would help the said scenarios? What are you trying to achieve and what are you OK to compromise live with?

I am curious to learn more about the real-world scenarios to validate and back my own hypothesis.

[ghahramani]

@ghahramani: When you get a chance, can you describe your use-case, and why the backpressure and being fully reactive would help the said scenarios? What are you trying to achieve and what are you OK to compromise live with?

I am curious to learn more about the real-world scenarios to validate and back my own hypothesis.

So basically, it is started from this Gitter conversation, feel free to read it if there is any confusion, I would be happy to clarify things

https://gitter.im/spring-cloud/spring-cloud-stream?at=608c38f32cc8c84d85013db4

[UgiR]

Out of curiosity, has any decision been made on this?

[sobychacko]

@UgiR Nothing concrete has been done towards this initiative yet.

[roger751]

Is there any progress in this issue? Seems to me like the most needed feature when using Project Reactor with Spring to consume from Apache Kafka. Isn't it? Can someone explain how the current situation even works? e.g. how do I currently consume reactively from Kafka with Spring Cloud Stream if I can't use Reactor Kafka?

[sobychacko]

A related issue for this one: https://github.com/spring-cloud/spring-cloud-function/issues/779

[garyrussell]

Also see this discussion https://github.com/reactor/reactor-kafka/issues/252#issuecomment-987309584

[roger751]

Replying to @artembilan's comment.

So, you need to decide for yourself if you stick with integration flow solution with whatever reactive capabilities are present there Are there any missing Reactive capabilities? I generally don't understand how does Spring Integration helps in this use case. My main question is:

Would you say that it is a good option to implement a Spring Integration channel adapter if you would want to use the Spring Cloud ecosystem and read with reactor kafka? And why?

And also as I've mentioned, I don't really understand - does Spring Integration has full reactive support, or do you consider it as a risky move?

Last thing - about the implementation with Spring Integration. Do you consider the implementation of a reactive channel adapter as of the same concept as the implementation of the futuristic spring-cloud-stream-binder-kafka? I mean, does the user's workaround for this case the same as the futuristic implementation plans?

[artembilan]

1. With the current Spring Cloud Stream Binder implementations there won't be technical difference if you just switch to plain Spring Integration flow solution. Just because current Spring Cloud Stream Binders are fully based on Spring Integration. Probably we can consider a conceptual difference since you won't be able to have binding with your custom solution: everything is going to be in your hands from that point.

2. Spring Integration is just a library, a set of components you can combine into a flow. In most cases a distributed async computation. So, it is wrong to judge it from the programming model perspective, since it is not a goal of this library. However with some convenient hooks it allows to implement a reactive solution. It is just might be the fact that not all the protocols Spring Integration has channel adapters for provide some reactive API. And even if they does, it doesn't mean we are able to catch up with everything what is there in the market. Like in this case with reactor-kafka: we just haven't implemented yet channel adapters for its abstractions.

3. No, Spring Integration channel adapter implementation has nothing to do with the binder abstraction. It might be the case that binder is based on the channel adapter, but might be not, like existing Binder for Apache Kafka Streams: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka#apache-kafka-streams-binder. And no, the workaround for your reactor-kafka move cannot be treated as what going to happen in the binder. You probably miss many Spring Cloud Stream bit just seeing only a channel adapter from the implementation perspective. There is a binder configuration to distinguish between others, there is bindings to handle, provisioning etc. The short and easy solution won't be able benefit from the rest of Spring Cloud Stream infrastructure. At least that's how I feel it...

[roger751]

@artembilan is there any example for the Reactor-Kafka with Spring Integration usage? It would be very helpful.

Also - @sobychacko is there any news about the progress \ plans for this issue? Can you tell what is needed to be implemented for this to be released?

[artembilan]

@roger751 ,

here is some sample about Reactor Kafka and Spring Integration interactions: https://github.com/artembilan/sandbox/tree/master/spring-integration-reactor-kafka.

I didn't add any README (yet), but behavior is simple:

1. Reactor Kafka producer used from the @ServiceActivator
2. Reactor Kafka consumer used from the IntegrationFlow
3. The test covers their behavior against embedded Kafka.

[roger751]

@artembilan looks awesome!!! Thank you very much!! Do you think it is worth being added to the spring-integration / spring-integration-kafka project as an official example/ official feature? Looks pretty important to me and in my opinion worth being an official channel adapter. What do you think?

And more related to the official feature request of this very important issue - @olegz @sobychacko @garyrussell do you think that a solution for writing a reactive Spring-Cloud-Stream application will be available anytime soon? Are there any plans of working on this? I think it's pretty misleading that the Spring ecosystem encourages devs to write Kafka consumer applications and reactive applications, but won't be able to offer a basic solution for doing both. And if the decision is to not use spring-cloud-stream, wouldn't it be more appropriate if there would be some blogs about the subject and its alternative solutions?

It is just very unclear right now what a developer who just want to consume and produce to Kafka with Reactor and Spring Boot should do, and it seems to me like the most obvious use case. Spring-Cloud-Stream is a great projects for synchronous applications, but if the Spring ecosystem so encouraging to develop Reactive applications, there should be an official and easy solution for this too.

[sobychacko]

@roger751 The plan is to rethink this idea of a reactive binder for Kafka as a new feature for the Spring Cloud Stream 4.0 roadmap. We will have more updates on this after the holidays.

[roger751]

@sobychacko ok. I'll be glad if you'll update us as soon as there will be progress. I think that 100% of the Reactor Kafka users are waiting for this, and probably lots of the Spring Cloud Stream users would be glad for such an integration between the two libraries as well. This integration is super important, and has a huge demand.

[lynch19]

+1

[kairurooj]

@sobychacko what's the timeline for Spring Cloud Stream 4.0?

[thepaep]

I would also like to know about progress in this area. Some blogs with updates will be great. If contributors' help is needed I would also like to know that.

[roger751]

@sobychacko for those of us that are waiting for this feature and need to implement data flow in the meantime, would you suggest implementing non Spring Cloud Stream using Function interfaces? I'm not sure what would be the easiest way of implementing applications so that the migration to Spring Cloud Stream 4.0.0 would be the easiest...

[markusherbert]

Definitely an important feature (in my opinion, that's the most important feature of this project). Any progress has been started? @sobychacko @artembilan @olegz

Isn't it ironic that SCSt supports all of Apache Kafka's clients (Consumer API and Streams API) except the one that has been implemented at Pivotal?

[sobychacko]

We are actively thinking through this right now and coming up with some design considerations for a new reactive binder. As soon as we have something shareable, we will do so. We hope to reach that state in a few weeks. Please stay tuned. Thank you!!

[roger751]

@sobychacko Great news! Looking forward to this. I also have a question about the conversion above. What’s the meaning of the Spring Cloud Sleuth support for “Reactive Spring Cloud Stream”, if there are no reactive binders? https://github.com/spring-cloud/spring-cloud-sleuth/commit/90417f1f074f41fed7b3bfb8e0c8c4e5a6274dd8

[fluxstirngchanle]

A very anticipated issue. Any timelines?

[rancherz]

Hey. I'd like to know if I understand correctly. I have a few questions:

1. At the current situation it is safe to use Reactor Kafka with Spring Integration but not with Spring Cloud Stream? I saw in the gitter conversation that there was an issue with SCSt. https://gitter.im/spring-cloud/spring-cloud-stream?at=608c38f32cc8c84d85013db4
2. I want to write a nio data pipeline that reads from Kafka, transforms (with usage of Out of Order Commits from Reactor-Kafka) and saves the output into Mongo. I wonder what is the most recommended approach to write such a pipeline. Should I use Spring Integration (and inject queues for each writer and transformer), or just us Reactor Kafka with a Processor class that serves as the "main"?

Any examples for applications that reads with Reactor Kafka and writes to some sink would be very appreciated (not just for Mongo).

@artembilan @ghahramani @garyrussell

[markusherbert]

Any updates? I'm willing to deploy a reactive Spring Integration application with Spring Cloud Dataflow, and this one blocks my migration from regular Spring Integration application, to Spring Cloud Stream.

[devlap2]

Does the existence of this issue mean that there is also not a real reactive support fr other binders in spring cloud stream (like azure eventhub and more)?

[garyrussell]

We've made some progress should have an initial PR for this soon.

@devlap2 Those binders are maintained by third parties.

[sobychacko]

Issue moved to spring-cloud/spring-cloud-stream #2293 via ZenHub