(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'