Make the consumer.close() graceful

[jeffwidman]

KafkaConsumer was originally written based on 0.9.0 Java code.

I noticed the Kafka release notes for 0.10.2 says:

Java consumer now shuts down gracefully. By default, the consumer waits up to 30 seconds to complete pending requests. A new close API with timeout has been added to KafkaConsumer to control the maximum wait time.

This should be investigated to see if it needs to be added to KafkaConsumer.

What we want to make sure is that the consumers issue LeaveGroupRequest in all situations so that the remaining consumers can rebalance faster rather than waiting for this closed consumer to timeout.

[dpkp]

Consumer timeouts were improved in 2.1 release, with the optional timeout_ms kwarg added to consumer.close() in 2.1.3