Closed adamsmith118 closed 3 months ago
KafkaBinder is not a reactive source or target, so it simply will not honor some of the reactive features (back pressure is one of them). That is by definition. It is imperative binder adapted to work with reactive functions - nothing more. If you want to rely on such features consider using KafkaStrams binder - a fully reactive binder
Tagging @sobychacko
@olegz This is using the Reactive Kafka Binder (i.e. the one in spring-cloud-stream-binder-kafka-reactive), not the old imperative one.
Are you saying this doesn't honour back pressure either?
To be clear - my issue is that the ReactorKafkaBinder in Spring Cloud Streams does not exhibit the same back pressure behaviour as when using a reactive KafkaReceiver directly (even though the Reactive Kafka Binder uses it internally).
No, i just wanted to clarify. With reactive binder we need to look, hence tagging @sobychacko
Ok - thanks.
I'll have a stab at debugging it now.
If I update FluxMessageChannel
to:
.publish(1).refCount()
instead of share()
(which specifies its own prefetch/queue defaults).publishOn(this.scheduler)
Then I get behaviour I want (if that's in any way useful?).
OK. I might see the value of the .publish(1).refCount()
, and FluxMessageChannel
might be fixed for that instead of Queues.SMALL_BUFFER_SIZE
.
But why to remove .publishOn(this.scheduler)
?
Typically used for fast publisher, slow consumer(s) scenarios.
according to its Javadocs.
Hi @artembilan,
Thanks for your response.
Whilst I see your point about publishOn()
I wonder if this choice should be left to the developer rather than being baked in to the framework? For example, I can easily add this into my consumer code if needed.
Open to suggestions...
Closed in favor of https://github.com/spring-projects/spring-integration/issues/9215. There is just nothing to do on the Spring Cloud Stream side. The fix in Spring Integration is going to be available as transitive dependency in the next release over here.
Thank you for your contribution!
Spring Integration 6.2.6-SNAPSHOT
(or 6.3.1-SNAPSHOT
) is available if you wish to give it a try in your project with Reactive Kafka Binder.
Keep in mind to override the spring-integration version in the application since the one that Spring Cloud Stream brings is what is currently available via Boot (which is not the snapshot version right now).
Expected behaviour
Parity in how back pressure is handled when using the SCS ReactorKafkaBinder and ReactorKafka directly.
If this isn't expected, guidance on how to achieve parity would be appreciated.
Actual behaviour
Consider two consumer implementations performing identical tasks; a simulation of some work followed by a WebClient call.
Using Reactor Kafka Directly
max.poll.records=1 for both, to keep things simple.
Using SCS Reactor Kafka Binder
In reactor kafka (example 1) we can see behaviour in-line with back pressure requirements. One record is emitted at a time. The consumer pauses as necessary.
In Spring Cloud Streams, using the ReactorKafkaBinder, this isn't the case.
Here, 100's of records are emitted to the sink:
This causes problems if a rebalance occurs during a period of heavy load as the pipeline can contain 100's of pending records.
We'd need to set an intolerably high maxDelayRebalance to get through them all or handle lots of duplicates.
Logs resembling the below are visible during a rebalance.
Presumably something in the binder/channel implementation is causing this?
Repro
See here for complete repro.
Requires a Kafka on localhost:9092 and a topic called "test".
Producer
class in test will send 100 messages.demo.reactor.DemoReactorApp
- Pure Reactor Kafka exampledemo.streams.DemoStreamsApplication
- Spring Cloud Streams exampleDEBUG logging has been enabled for the
ConsumerEventLoop
for emit visibility.Environment details
Java 21 Boot 3.2.2 SCS: 4.1.0 Reactor Kafka: 1.3.22
Loosely related issue here.