Execution of RecordsStreamProducer.closeConnections() is susceptible to race condition

The closeConnections() method is called if and only if the method start() was executed successfully setting replicationStream. The problem is that the method closes also replicationConnection that is created in the constructor. The ent result is that the resources might not be cleaned if the startup sequence has not been completed. This can be issue especially with cluster rebalancing.