lib/kafka/client.rb
# 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.
# @param refresh_topic_interval [Integer] 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
# @param interceptors [Array<Object>] a list of consumer interceptors that implement
#   `call(Kafka::FetchedBatch)`.
# @param assignment_strategy [Object] a partition assignment strategy that
#   implements `protocol_type()`, `user_data()`, and `assign(members:, partitions:)`
# @return [Consumer]
Creates a new Kafka consumer.
Creates a new Kafka consumer .
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,
      })
      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
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 , } ) 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
Creates a new Kafka consumer.
Creates a new Kafka consumer .
# The Kafka protocol expects the retention time to be in ms.
None
Consumer
group_id
the id of the group that the consumer should join.
the id of the group that the consumer should join .
session_timeout
the number of seconds after which, if a client hasn't contacted the Kafka cluster, it will be kicked out of the group.
the number of seconds after which if a client hasn ' t contacted the Kafka cluster it will be kicked out of the group .
rebalance_timeout
<not_specific>
offset_commit_interval
the interval between offset commits, in seconds.
the interval between offset commits in seconds .
offset_commit_threshold
the number of messages that can be processed before their offsets are committed. If zero, offset commits are not triggered by message processing.
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
the interval between heartbeats; must be less than the session window.
the interval between heartbeats ; must be less than the session window .
offset_retention_time
the time period that committed offsets will be retained, in seconds. Defaults to the broker setting.
the time period that committed offsets will be retained in seconds . Defaults to the broker setting .
fetcher_max_queue_size
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.
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
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
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
a list of consumer interceptors that implement `call(Kafka::FetchedBatch)`.
a list of consumer interceptors that implement ` call ( Kafka :: FetchedBatch ) ` .
assignment_strategy
a partition assignment strategy that implements `protocol_type()`, `user_data()`, and `assign(members:, partitions:)`
a partition assignment strategy that implements ` protocol_type () ` ` user_data () ` and ` assign ( members : partitions : ) `