def consumer(
        group_id:,
        session_timeout: 30,
        offset_commit_interval: 10,
        offset_commit_threshold: 0,
        heartbeat_interval: 10,
        offset_retention_time: nil,
        fetcher_max_queue_size: 100
    )
      cluster = initialize_cluster

      instrumenter = DecoratingInstrumenter.new(@instrumenter, {
        group_id: group_id,
      })

      # The Kafka protocol expects the retention time to be in ms.
      retention_time = (offset_retention_time && offset_retention_time * 1_000) || -1

      group = ConsumerGroup.new(
        cluster: cluster,
        logger: @logger,
        group_id: group_id,
        session_timeout: session_timeout,
        retention_time: retention_time,
        instrumenter: instrumenter,
      )

      fetcher = Fetcher.new(
        cluster: initialize_cluster,
        group: group,
        logger: @logger,
        instrumenter: instrumenter,
        max_queue_size: fetcher_max_queue_size
      )

      offset_manager = OffsetManager.new(
        cluster: cluster,
        group: group,
        fetcher: fetcher,
        logger: @logger,
        commit_interval: offset_commit_interval,
        commit_threshold: offset_commit_threshold,
        offset_retention_time: offset_retention_time
      )

      heartbeat = Heartbeat.new(
        group: group,
        interval: heartbeat_interval,
        instrumenter: instrumenter
      )

      Consumer.new(
        cluster: cluster,
        logger: @logger,
        instrumenter: instrumenter,
        group: group,
        offset_manager: offset_manager,
        fetcher: fetcher,
        session_timeout: session_timeout,
        heartbeat: heartbeat,
      )
    end
def consumer(
        group_id:,
        session_timeout: 30,
        offset_commit_interval: 10,
        offset_commit_threshold: 0,
        heartbeat_interval: 10,
        offset_retention_time: nil,
        fetcher_max_queue_size: 100
    )
      cluster = initialize_cluster

      instrumenter = DecoratingInstrumenter.new(@instrumenter, {
        group_id: group_id,
      })

      # The Kafka protocol expects the retention time to be in ms.
      retention_time = (offset_retention_time && offset_retention_time * 1_000) || -1

      group = ConsumerGroup.new(
        cluster: cluster,
        logger: @logger,
        group_id: group_id,
        session_timeout: session_timeout,
        retention_time: retention_time,
        instrumenter: instrumenter,
      )

      fetcher = Fetcher.new(
        cluster: initialize_cluster,
        group: group,
        logger: @logger,
        instrumenter: instrumenter,
        max_queue_size: fetcher_max_queue_size
      )

      offset_manager = OffsetManager.new(
        cluster: cluster,
        group: group,
        fetcher: fetcher,
        logger: @logger,
        commit_interval: offset_commit_interval,
        commit_threshold: offset_commit_threshold,
        offset_retention_time: offset_retention_time
      )

      heartbeat = Heartbeat.new(
        group: group,
        interval: heartbeat_interval,
        instrumenter: instrumenter
      )

      Consumer.new(
        cluster: cluster,
        logger: @logger,
        instrumenter: instrumenter,
        group: group,
        offset_manager: offset_manager,
        fetcher: fetcher,
        session_timeout: session_timeout,
        heartbeat: heartbeat,
      )
    end
Creates a new Kafka consumer.

 @param group_id [String] the id of the group that the consumer should join.
 @param session_timeout [Integer] the number of seconds after which, if a client
   hasn't contacted the Kafka cluster, it will be kicked out of the group.
 @param offset_commit_interval [Integer] the interval between offset commits,
   in seconds.
 @param offset_commit_threshold [Integer] the number of messages that can be
   processed before their offsets are committed. If zero, offset commits are
   not triggered by message processing.
 @param heartbeat_interval [Integer] the interval between heartbeats; must be less
   than the session window.
 @param offset_retention_time [Integer] the time period that committed
   offsets will be retained, in seconds. Defaults to the broker setting.
 @param fetcher_max_queue_size [Integer] max number of items in the fetch queue that
   are stored for further processing. Note, that each item in the queue represents a
   response from a single broker.
 @return [Consumer]