Class: Racecar::RebalanceListener

Inherits:
Object
  • Object
show all
Defined in:
lib/racecar/rebalance_listener.rb

Defined Under Namespace

Classes: Event

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(consumer_class, instrumenter) ⇒ RebalanceListener

Returns a new instance of RebalanceListener.



3
4
5
6
7
 
# File 'lib/racecar/rebalance_listener.rb', line 3

def initialize(consumer_class, instrumenter)
  @consumer_class = consumer_class
  @instrumenter = instrumenter
  @rdkafka_consumer = nil
end

Instance Attribute Details

#rdkafka_consumer=(value) ⇒ Object

Sets the attribute rdkafka_consumer

Parameters:

  • value —

    the value to set the attribute rdkafka_consumer to.



9
10
11
 
# File 'lib/racecar/rebalance_listener.rb', line 9

def rdkafka_consumer=(value)
  @rdkafka_consumer = value
end

Instance Method Details

#on_partitions_assigned(rdkafka_topic_partition_list) ⇒ Object 



14
15
16
17
18
19
20
 
# File 'lib/racecar/rebalance_listener.rb', line 14

def on_partitions_assigned(rdkafka_topic_partition_list)
  event = Event.new(rdkafka_consumer: rdkafka_consumer, rdkafka_topic_partition_list: rdkafka_topic_partition_list)

  instrument("partitions_assigned", partitions: event.partition_numbers) do
    consumer_class.on_partitions_assigned(event)
  end
end

#on_partitions_revoked(rdkafka_topic_partition_list) ⇒ Object 



22
23
24
25
26
27
28
 
# File 'lib/racecar/rebalance_listener.rb', line 22

def on_partitions_revoked(rdkafka_topic_partition_list)
  event = Event.new(rdkafka_consumer: rdkafka_consumer, rdkafka_topic_partition_list: rdkafka_topic_partition_list)

  instrument("partitions_revoked", partitions: event.partition_numbers) do
    consumer_class.on_partitions_revoked(event)
  end
end