Clean up connections inside BrokerPool
Clean up BrokerPool inside consumer_for_partition