Hey folks, during the production use of the library I've noticed that whenever an error occurs (due to improper binder configuration, such as invalid topic name, or an authorization error on a broker, for example) when trying to fetch partition information for the topic on binding a Kafka Producer to KafkaMessageChannelBinder, it leads to a resource leak due to continous attempts to retry the initialization and subsequent recreation of Producer instances.

Which basically means the Producer instances (and whatever they reference), Producer IO thread count, open file descriptors, etc. keep growing indefinitely until resource exhaustion occurs - whether it's platform threads, or heap space, or a file descriptor limit. It can also cause a DDOS on the broker side due to leaking producers taking up TCP connections and never releasing them - we've actually experienced this first hand when 18 kubernetes pods suddenly began spamming our kafka broker with a large count of TCP connections that were kept alive forever.

After some time spent debugging the issue, I found out the root cause and felt that the solution is so glaringly obvious that I didn't bother creating a POC project and went straight for PR with a fix.

Please note that the issue isn't reproduced when fetching partition info for consumers in KafkaTopicProvisioner#getPartitionInfoForConsumer due to a proper use of try-with-resources.

Thanks!

@akenra Thanks for the PR! Could you add your name as an author to the provisioner class you modified?

@akenra Thanks for the PR! Could you add your name as an author to the provisioner class you modified?

Of course! Done now.

@akenra Merged the PR and back-ported to 4.2.x and 4.1.x branches. Thanks again!

Here is the release schedule: https://github.com/spring-cloud/spring-cloud-release/milestones?direction=asc&sort=due_date&state=open