make changes to be more fault tolerant: clean up connections, brokers, failed_messages

- add integration tests for sync producer
- add integration tests for async producer w. leadership election
- use log.exception