Method: Kafka::Client#consumer

Defined in:

lib/kafka/client.rb

#consumer(group_id:, session_timeout: 30, rebalance_timeout: 60, offset_commit_interval: 10, offset_commit_threshold: 0, heartbeat_interval: 10, offset_retention_time: nil, fetcher_max_queue_size: 100, refresh_topic_interval: 0, interceptors: [], assignment_strategy: nil) ⇒ Consumer

Creates a new Kafka consumer.

Parameters:

• group_id (String) —

  the id of the group that the consumer should join.

• session_timeout (Integer) (defaults to: 30) —

  the number of seconds after which, if a client hasn't contacted the Kafka cluster, it will be kicked out of the group.

• offset_commit_interval (Integer) (defaults to: 10) —

  the interval between offset commits, in seconds.

• offset_commit_threshold (Integer) (defaults to: 0) —

  the number of messages that can be processed before their offsets are committed. If zero, offset commits are not triggered by message processing.

• heartbeat_interval (Integer) (defaults to: 10) —

  the interval between heartbeats; must be less than the session window.

• offset_retention_time (Integer) (defaults to: nil) —

  the time period that committed offsets will be retained, in seconds. Defaults to the broker setting.

• fetcher_max_queue_size (Integer) (defaults to: 100) —

  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.

• refresh_topic_interval (Integer) (defaults to: 0) —

  interval of refreshing the topic list. If it is 0, the topic list won't be refreshed (default) If it is n (n > 0), the topic list will be refreshed every n seconds

• interceptors (Array<Object>) (defaults to: []) —

  a list of consumer interceptors that implement call(Kafka::FetchedBatch).

• assignment_strategy (Object) (defaults to: nil) —

  a partition assignment strategy that implements protocol_type(), user_data(), and assign(members:, partitions:)

Returns:

• (Consumer)

# File 'lib/kafka/client.rb', line 380

def consumer(
    group_id:,
    session_timeout: 30,
    rebalance_timeout: 60,
    offset_commit_interval: 10,
    offset_commit_threshold: 0,
    heartbeat_interval: 10,
    offset_retention_time: nil,
    fetcher_max_queue_size: 100,
    refresh_topic_interval: 0,
    interceptors: [],
    assignment_strategy: nil
)
  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,
    rebalance_timeout: rebalance_timeout,
    retention_time: retention_time,
    instrumenter: instrumenter,
    assignment_strategy: assignment_strategy
  )

  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,
    refresh_topic_interval: refresh_topic_interval,
    interceptors: interceptors
  )
end