(Reactive) Bindings do not recover from application restart

[kschlesselmann]

Describe the issue
It seems that functions (especially producers) do not recover from application restarts. We were trying to use the auto property reload features from Spring AWS and after a context restart our functions are broken.

I've created a minimal sample to show the issue.

To Reproduce
Steps to reproduce the behavior:

1. Start kafka via docker compose up
2. Start the application
3. Watch for successfully consumed messages (It just prints 1, 2, 3, ...)
4. Restart the application via http POST :8080/actuator/restart
5. Now the stream functions are broken

Expected behavior
The application should restart and the bindings should be fine again.

[FlxGschlr]

I've played around with the sample and it seems like the the producer is not shutdown correctly on restart. Add a print to the producer bean and you'll see that there are two producers after the /restart call. One works as expected and one throws the following error:

org.springframework.messaging.MessageDeliveryException: failed to send Message to channel 'produce_integrationflow.channel#0'

[olegz]

@kschlesselmann please give it a try and let me know

[olegz]

We are also tracking a potential problem in Spring Integration (SI) where the integration flow which represents the start of the Supplier flow is not stopped albeit being Lifecycle.
In other words, while this patch addresses the immediate issue it may actually be reverted and polished in the future depending in how SI addresses the issue.

@artembilan, can you please inlclude the link to the corresponding SI issue?

[artembilan]

Related Spring Integration issue: spring-projects/spring-integration#9999
(nice number though! 😄)

[kschlesselmann]

@olegz In my minimal producer it looks fine. If I try it in our more complex production scenario (where I change credentials for an OAuth2 Web Client) the application doesn't really recover. I think that's another problem though and maybe the SCSt side is alright now 👍🏻

[artembilan]

So, the fix is in Spring Integration.
Technically, the workaround provided here in Spring Cloud Stream could be reverted, since the FluxMessageChannel is now a Lifecycle called from the ApplicaitonContext.start() and stop().

[artembilan]

Looked into the current code of Spring Cloud Stream and feels like my idea about removal is not OK yet.
That fix in Spring Integration is still needed, but looks like the whole IntegrationFlow for supplier-based binding is recreated on application restart.

@olegz , let's chat when you come back!