Class: Kafka::ConsumerGroup

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

Defined Under Namespace

Classes: Assignor

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(cluster:, logger:, group_id:, session_timeout:, rebalance_timeout:, retention_time:, instrumenter:, assignment_strategy:) ⇒ ConsumerGroup

Returns a new instance of ConsumerGroup.



11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
 
# File 'lib/kafka/consumer_group.rb', line 11

def initialize(cluster:, logger:, group_id:, session_timeout:, rebalance_timeout:, retention_time:, instrumenter:, assignment_strategy:)
  @cluster = cluster
  @logger = TaggedLogger.new(logger)
  @group_id = group_id
  @session_timeout = session_timeout
  @rebalance_timeout = rebalance_timeout
  @instrumenter = instrumenter
  @member_id = ""
  @generation_id = nil
  @members = {}
  @topics = Set.new
  @assigned_partitions = {}
  @assignor = Assignor.new(
    cluster: cluster,
    strategy: assignment_strategy || RoundRobinAssignmentStrategy.new
  )
  @retention_time = retention_time
end

Instance Attribute Details

#assigned_partitions ⇒ Object (readonly)

Returns the value of attribute assigned_partitions.



9
10
11
 
# File 'lib/kafka/consumer_group.rb', line 9

def assigned_partitions
  @assigned_partitions
end

#generation_id ⇒ Object (readonly)

Returns the value of attribute generation_id.



9
10
11
 
# File 'lib/kafka/consumer_group.rb', line 9

def generation_id
  @generation_id
end

#group_id ⇒ Object (readonly)

Returns the value of attribute group_id.



9
10
11
 
# File 'lib/kafka/consumer_group.rb', line 9

def group_id
  @group_id
end

Instance Method Details

#assigned_to?(topic, partition) ⇒ Boolean

Returns:

  • (Boolean) 


39
40
41
 
# File 'lib/kafka/consumer_group.rb', line 39

def assigned_to?(topic, partition)
  subscribed_partitions.fetch(topic, []).include?(partition)
end

#commit_offsets(offsets) ⇒ Object 



86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
 
# File 'lib/kafka/consumer_group.rb', line 86

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

  response.topics.each do |topic, partitions|
    partitions.each do |partition, error_code|
      Protocol.handle_error(error_code)
    end
  end
rescue Kafka::Error => e
  @logger.error "Error committing offsets: #{e}"
  raise OffsetCommitError, e
end

#fetch_offsets ⇒ Object 



79
80
81
82
83
84
 
# File 'lib/kafka/consumer_group.rb', line 79

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

#heartbeat ⇒ Object 



105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
 
# File 'lib/kafka/consumer_group.rb', line 105

def heartbeat
  @logger.debug "Sending heartbeat..."

  @instrumenter.instrument('heartbeat.consumer',
                           group_id: @group_id,
                           topic_partitions: @assigned_partitions) do

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

    Protocol.handle_error(response.error_code)
  end
rescue ConnectionError, UnknownMemberId, IllegalGeneration => e
  @logger.error "Error sending heartbeat: #{e}"
  raise HeartbeatError, e
rescue RebalanceInProgress => e
  @logger.warn "Error sending heartbeat: #{e}"
  raise HeartbeatError, e
rescue NotCoordinatorForGroup
  @logger.error "Failed to find coordinator for group `#{@group_id}`; retrying..."
  sleep 1
  @coordinator = nil
  retry
end

#join ⇒ Object 



47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/kafka/consumer_group.rb', line 47

def join
  if @topics.empty?
    raise Kafka::Error, "Cannot join group without at least one topic subscription"
  end

  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..."
  sleep 1
  @cluster.mark_as_stale!
  @coordinator = nil
  retry
end

#leave ⇒ Object 



67
68
69
70
71
72
73
74
75
76
77
 
# File 'lib/kafka/consumer_group.rb', line 67

def leave
  @logger.info "Leaving group `#{@group_id}`"

  # Having a generation id indicates that we're a member of the group.
  @generation_id = nil

  @instrumenter.instrument("leave_group.consumer", group_id: @group_id) do
    coordinator.leave_group(group_id: @group_id, member_id: @member_id)
  end
rescue ConnectionError
end

#member? ⇒ Boolean

Returns:

  • (Boolean) 


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

def member?
  !@generation_id.nil?
end

#subscribe(topic) ⇒ Object 



30
31
32
33
 
# File 'lib/kafka/consumer_group.rb', line 30

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

#subscribed_partitions ⇒ Object 



35
36
37
 
# File 'lib/kafka/consumer_group.rb', line 35

def subscribed_partitions
  @assigned_partitions.select { |topic, _| @topics.include?(topic) }
end

#to_s ⇒ Object 



133
134
135
136
137
138
139
140
 
# File 'lib/kafka/consumer_group.rb', line 133

def to_s
  "[#{@group_id}] {" + assigned_partitions.map { |topic, partitions|
    partition_str = partitions.size > 5 ?
                      "#{partitions[0..4].join(', ')}..." :
                      partitions.join(', ')
    "#{topic}: #{partition_str}"
  }.join('; ') + '}:'
end