Class: Karafka::Instrumentation::Callbacks::ConsumerGroups::Rebalance

Inherits:
Object
  • Object
show all
Defined in:
lib/karafka/instrumentation/callbacks/consumer_groups/rebalance.rb

Overview

Callback that connects to the librdkafka rebalance callback and converts those events into our internal events

Instance Method Summary collapse

Constructor Details

#initialize(subscription_group, client_id) ⇒ Rebalance

Returns a new instance of Rebalance.

Parameters:

  • subscription_group (Karafka::Routes::SubscriptionGroup) —

    subscription group for which we want to manage rebalances

  • client_id (String) —

    id of the client managing this rebalance



18
19
20
21
 
# File 'lib/karafka/instrumentation/callbacks/consumer_groups/rebalance.rb', line 18

def initialize(subscription_group, client_id)
  @subscription_group = subscription_group
  @client_id = client_id
end

Instance Method Details

#on_partitions_assign(tpl) ⇒ Object

Publishes an event that partitions are going to be assigned

Parameters:

  • tpl (Rdkafka::Consumer::TopicPartitionList) 


34
35
36
 
# File 'lib/karafka/instrumentation/callbacks/consumer_groups/rebalance.rb', line 34

def on_partitions_assign(tpl)
  instrument("partitions_assign", tpl)
end

#on_partitions_assigned(tpl) ⇒ Object

Publishes an event that partitions were assigned.

Parameters:

  • tpl (Rdkafka::Consumer::TopicPartitionList) 


49
50
51
 
# File 'lib/karafka/instrumentation/callbacks/consumer_groups/rebalance.rb', line 49

def on_partitions_assigned(tpl)
  instrument("partitions_assigned", tpl)
end

#on_partitions_revoke(tpl) ⇒ Object

Publishes an event that partitions are going to be revoked. At this stage we can still commit offsets, etc.

Parameters:

  • tpl (Rdkafka::Consumer::TopicPartitionList) 


27
28
29
 
# File 'lib/karafka/instrumentation/callbacks/consumer_groups/rebalance.rb', line 27

def on_partitions_revoke(tpl)
  instrument("partitions_revoke", tpl)
end

#on_partitions_revoked(tpl) ⇒ Object

Publishes an event that partitions were revoked. This is after we’ve lost them, so no option to commit offsets.

Parameters:

  • tpl (Rdkafka::Consumer::TopicPartitionList) 


42
43
44
 
# File 'lib/karafka/instrumentation/callbacks/consumer_groups/rebalance.rb', line 42

def on_partitions_revoked(tpl)
  instrument("partitions_revoked", tpl)
end