Close message channel issue

[omercelikceng]

Resolves #2869

[olegz]

@omercelikceng The issue #2869 marked as closed. Can you please explain what you are adding with this PR?

[omercelikceng]

@omercelikceng The issue #2869 marked as closed. Can you please explain what you are adding with this PR?

Hi @olegz . I'd like to explain using the tests I've written. Assume we want to produce data by providing a binder to a dynamic destination.

In this example, when data is sent, it is put into the channelCache map within the StreamBridge class under the key "kafka1:dynamicTopic". The relevant code is shown in the image below.

Additionally, the this.bindingService.bindProducer(..) method is also invoked here. Let’s examine this method. When this method is called, the data is put into the producerBindings map within the BindingService class under the key "dynamicTopic". The relevant code is shown in the image below.

Moreover, all classes derived from AbstractBindableProxyFactory populate the outputHolders map, where the key is solely the output name. These classes also have the capability to invoke the unbindProducer method.

Subsequently, when the application is shutting down, the destroy method of the StreamBridge class is executed. All relevant values in the channelCache map are unbound by calling the unbindProducers method there. (via channelCache.keySet().forEach(bindingService::unbindProducers);). However, since the keys in channelCache map differ from those in producerBindings map, the unbind operation is not performed.

In other words, the key values in these maps are not identical. To resolve this issue during the unbind process, I made several enhancements to correct the inconsistencies.

If you run the tests I have written with the previous code, you will see that the error occurs.