'Close the consumer, waiting indefinitely for any needed cleanup.
Keyword Arguments:
autocommit (bool): If auto-commit is configured for this consumer,
this optional flag causes the consumer to attempt to commit any
pending consumed offsets prior to close. Default: True'

def close(self, autocommit=True):

    if self._closed:
        return
    log.debug('Closing    the    KafkaConsumer.')
    self._closed = True
    self._coordinator.close(autocommit=autocommit)
    self._metrics.close()
    self._client.close()
    try:
        self.config['key_deserializer'].close()
    except AttributeError:
        pass
    try:
        self.config['value_deserializer'].close()
    except AttributeError:
        pass
    log.debug('The    KafkaConsumer    has    closed.')
