docs: def configure(self, **configs):
          """Configure the consumer instance
  
          Configuration settings can be passed to constructor,
          otherwise defaults will be used:
  
          Keyword Arguments:
              bootstrap_servers (list): List of initial broker nodes the consumer
                  should contact to bootstrap initial cluster metadata.  This does
                  not have to be the full node list.  It just needs to have at
                  least one broker that will respond to a Metadata API Request.
              client_id (str): a unique name for this client.  Defaults to
                  'kafka.consumer.kafka'.
              group_id (str): the name of the consumer group to join,
                  Offsets are fetched / committed to this group name.
              fetch_message_max_bytes (int, optional): Maximum bytes for each
                  topic/partition fetch request.  Defaults to 1024*1024.
              fetch_min_bytes (int, optional): Minimum amount of data the server
                  should return for a fetch request, otherwise wait up to
                  fetch_wait_max_ms for more data to accumulate.  Defaults to 1.
              fetch_wait_max_ms (int, optional): Maximum time for the server to
                  block waiting for fetch_min_bytes messages to accumulate.
                  Defaults to 100.
              refresh_leader_backoff_ms (int, optional): Milliseconds to backoff
                  when refreshing metadata on errors (subject to random jitter).
                  Defaults to 200.
              socket_timeout_ms (int, optional): TCP socket timeout in
                  milliseconds.  Defaults to 30*1000.
              auto_offset_reset (str, optional): A policy for resetting offsets on
                  OffsetOutOfRange errors. 'smallest' will move to the oldest
                  available message, 'largest' will move to the most recent.  Any
                  ofther value will raise the exception.  Defaults to 'largest'.
              deserializer_class (callable, optional):  Any callable that takes a
                  raw message value and returns a deserialized value.  Defaults to
                   lambda msg: msg.
              auto_commit_enable (bool, optional): Enabling auto-commit will cause
                  the KafkaConsumer to periodically commit offsets without an
                  explicit call to commit().  Defaults to False.
              auto_commit_interval_ms (int, optional):  If auto_commit_enabled,
                  the milliseconds between automatic offset commits.  Defaults to
                  60 * 1000.
              auto_commit_interval_messages (int, optional): If
                  auto_commit_enabled, a number of messages consumed between
                  automatic offset commits.  Defaults to None (disabled).
              consumer_timeout_ms (int, optional): number of millisecond to throw
                  a timeout exception to the consumer if no message is available
                  for consumption.  Defaults to -1 (dont throw exception).
  
          Configuration parameters are described in more detail at
          http://kafka.apache.org/documentation.html#highlevelconsumerapi
          """
          configs = self._deprecate_configs(**configs)
          self._config = {}
          for key in self.DEFAULT_CONFIG:
              self._config[key] = configs.pop(key, self.DEFAULT_CONFIG[key])
  
          if configs:
              raise KafkaConfigurationError('Unknown configuration key(s): ' +
                                            str(list(configs.keys())))
  
          if self._config['auto_commit_enable']:
              if not self._config['group_id']:
                  raise KafkaConfigurationError(
                      'KafkaConsumer configured to auto-commit '
                      'without required consumer group (group_id)'
                  )
  
          # Check auto-commit configuration
          if self._config['auto_commit_enable']:
              logger.info("Configuring consumer to auto-commit offsets")
              self._reset_auto_commit()
  
          if not self._config['bootstrap_servers']:
              raise KafkaConfigurationError(
                  'bootstrap_servers required to configure KafkaConsumer'
              )
  
          self._client = KafkaClient(
              self._config['bootstrap_servers'],
              client_id=self._config['client_id'],
              timeout=(self._config['socket_timeout_ms'] / 1000.0)
          )