Class: Kafka::ConsumerGroup

Inherits:
Object
  • Object
show all
Defined in:
lib/kafka/consumer_group.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(cluster:, logger:, group_id:, session_timeout:) ⇒ ConsumerGroup

Returns a new instance of ConsumerGroup.



8
9
10
11
12
13
14
15
16
17
18
19
 
# File 'lib/kafka/consumer_group.rb', line 8

def initialize(cluster:, logger:, group_id:, session_timeout:)
  @cluster = cluster
  @logger = logger
  @group_id = group_id
  @session_timeout = session_timeout
  @member_id = ""
  @generation_id = nil
  @members = {}
  @topics = Set.new
  @assigned_partitions = {}
  @assignment_strategy = RoundRobinAssignmentStrategy.new(cluster: @cluster)
end

Instance Attribute Details

#assigned_partitions ⇒ Object (readonly)

Returns the value of attribute assigned_partitions.



6
7
8
 
# File 'lib/kafka/consumer_group.rb', line 6

def assigned_partitions
  @assigned_partitions
end

Instance Method Details

#commit_offsets(offsets) ⇒ Object 



56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
 
# File 'lib/kafka/consumer_group.rb', line 56

def commit_offsets(offsets)
  response = coordinator.commit_offsets(
    group_id: @group_id,
    member_id: @member_id,
    generation_id: @generation_id,
    offsets: offsets,
  )

  response.topics.each do |topic, partitions|
    partitions.each do |partition, error_code|
      Protocol.handle_error(error_code)
    end
  end
rescue UnknownMemberId
  @logger.error "Kicked out of group; rejoining"
  join
  retry
rescue IllegalGeneration
  @logger.error "Illegal generation #{@generation_id}; rejoining group"
  join
  retry
end

#fetch_offsets ⇒ Object 



49
50
51
52
53
54
 
# File 'lib/kafka/consumer_group.rb', line 49

def fetch_offsets
  coordinator.fetch_offsets(
    group_id: @group_id,
    topics: @assigned_partitions,
  )
end

#heartbeat ⇒ Object 



79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/kafka/consumer_group.rb', line 79

def heartbeat
  @logger.info "[#{@member_id}] Sending heartbeat..."

  response = coordinator.heartbeat(
    group_id: @group_id,
    generation_id: @generation_id,
    member_id: @member_id,
  )

  Protocol.handle_error(response.error_code)
rescue ConnectionError => e
  @logger.error "Connection error while sending heartbeat; rejoining"
  join
rescue UnknownMemberId
  @logger.error "Kicked out of group; rejoining"
  join
rescue RebalanceInProgress
  @logger.error "Group is rebalancing; rejoining"
  join
end

#join ⇒ Object 



30
31
32
33
34
35
36
37
38
39
40
41
42
 
# File 'lib/kafka/consumer_group.rb', line 30

def join
  join_group
  synchronize
rescue NotCoordinatorForGroup
  @logger.error "Failed to find coordinator for group `#{@group_id}`; retrying..."
  sleep 1
  @coordinator = nil
  retry
rescue ConnectionError
  @logger.error "Connection error while trying to join group `#{@group_id}`; retrying..."
  @coordinator = nil
  retry
end

#leave ⇒ Object 



44
45
46
47
 
# File 'lib/kafka/consumer_group.rb', line 44

def leave
  @logger.info "[#{@member_id}] Leaving group `#{@group_id}`"
  coordinator.leave_group(group_id: @group_id, member_id: @member_id)
end

#member? ⇒ Boolean

Returns:

  • (Boolean) 


26
27
28
 
# File 'lib/kafka/consumer_group.rb', line 26

def member?
  !@generation_id.nil?
end

#subscribe(topic) ⇒ Object 



21
22
23
24
 
# File 'lib/kafka/consumer_group.rb', line 21

def subscribe(topic)
  @topics.add(topic)
  @cluster.add_target_topics([topic])
end