Improved error handling in kafka_consumer

If the kafka_consumer plugin encounters an error fetching consumer offsets
(like for a partition which is configured but no longer exists), it can
start to allocate threads and never close them oout.  I added additional
exception handling around fetching both consumer and broker offsets so
that the plugin will log errors, but not fail altogether, and not cause
runaway thread usage.

Change-Id: I03e6f7cc3ecd787d7b4ad54b74136540607a465a