Class: Kafka::Consumer
- Inherits:
-
Object
- Object
- Kafka::Consumer
- Defined in:
- lib/kafka/consumer.rb
Overview
A client that consumes messages from a Kafka cluster in coordination with other clients.
A Consumer subscribes to one or more Kafka topics; all consumers with the same group id then agree on who should read from the individual topic partitions. When group members join or leave, the group synchronizes, making sure that all partitions are assigned to a single member, and that all members have some partitions to read from.
Example
A simple producer that simply writes the messages it consumes to the console.
require "kafka"
kafka = Kafka.new(["kafka1:9092", "kafka2:9092"])
# Create a new Consumer instance in the group `my-group`:
consumer = kafka.consumer(group_id: "my-group")
# Subscribe to a Kafka topic:
consumer.subscribe("messages")
# Loop forever, reading in messages from all topics that have been
# subscribed to.
consumer. do ||
puts .topic
puts .partition
puts .key
puts .value
puts .offset
end
Instance Method Summary collapse
- #commit_offsets ⇒ Object
-
#each_batch(min_bytes: 1, max_bytes: 10485760, max_wait_time: 1, automatically_mark_as_processed: true) {|batch| ... } ⇒ nil
Fetches and enumerates the messages in the topics that the consumer group subscribes to.
-
#each_message(min_bytes: 1, max_bytes: 10485760, max_wait_time: 1, automatically_mark_as_processed: true) {|message| ... } ⇒ nil
Fetches and enumerates the messages in the topics that the consumer group subscribes to.
-
#initialize(cluster:, logger:, instrumenter:, group:, fetcher:, offset_manager:, session_timeout:, heartbeat:) ⇒ Consumer
constructor
A new instance of Consumer.
- #mark_message_as_processed(message) ⇒ Object
-
#pause(topic, partition, timeout: nil, max_timeout: nil, exponential_backoff: false) ⇒ nil
Pause processing of a specific topic partition.
-
#paused?(topic, partition) ⇒ Boolean
Whether the topic partition is currently paused.
-
#resume(topic, partition) ⇒ nil
Resume processing of a topic partition.
-
#seek(topic, partition, offset) ⇒ nil
Move the consumer's position in a topic partition to the specified offset.
- #send_heartbeat_if_necessary ⇒ Object
-
#stop ⇒ nil
Stop the consumer.
-
#subscribe(topic, default_offset: nil, start_from_beginning: true, max_bytes_per_partition: 1048576) ⇒ nil
Subscribes the consumer to a topic.
Constructor Details
#initialize(cluster:, logger:, instrumenter:, group:, fetcher:, offset_manager:, session_timeout:, heartbeat:) ⇒ Consumer
Returns a new instance of Consumer.
46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 |
# File 'lib/kafka/consumer.rb', line 46 def initialize(cluster:, logger:, instrumenter:, group:, fetcher:, offset_manager:, session_timeout:, heartbeat:) @cluster = cluster @logger = logger @instrumenter = instrumenter @group = group @offset_manager = offset_manager @session_timeout = session_timeout @fetcher = fetcher @heartbeat = heartbeat @pauses = Hash.new {|h, k| h[k] = Hash.new {|h2, k2| h2[k2] = Pause.new } } # Whether or not the consumer is currently consuming messages. @running = false # Hash containing offsets for each topic and partition that has the # automatically_mark_as_processed feature disabled. Offset manager is only active # when everything is suppose to happen automatically. Otherwise we need to keep track of the # offset manually in memory for all the time # The key structure for this equals an array with topic and partition [topic, partition] # The value is equal to the offset of the last message we've received # @note It won't be updated in case user marks message as processed, because for the case # when user commits message other than last in a batch, this would make ruby-kafka refetch # some already consumed messages @current_offsets = Hash.new { |h, k| h[k] = {} } end |
Instance Method Details
#commit_offsets ⇒ Object
357 358 359 |
# File 'lib/kafka/consumer.rb', line 357 def commit_offsets @offset_manager.commit_offsets end |
#each_batch(min_bytes: 1, max_bytes: 10485760, max_wait_time: 1, automatically_mark_as_processed: true) {|batch| ... } ⇒ nil
Fetches and enumerates the messages in the topics that the consumer group subscribes to.
Each batch of messages is yielded to the provided block. If the block returns without raising an exception, the batch will be considered successfully processed. At regular intervals the offset of the most recent successfully processed message batch in each partition will be committed to the Kafka offset store. If the consumer crashes or leaves the group, the group member that is tasked with taking over processing of these partitions will resume at the last committed offsets.
280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 |
# File 'lib/kafka/consumer.rb', line 280 def each_batch(min_bytes: 1, max_bytes: 10485760, max_wait_time: 1, automatically_mark_as_processed: true) @fetcher.configure( min_bytes: min_bytes, max_bytes: max_bytes, max_wait_time: max_wait_time, ) consumer_loop do batches = fetch_batches batches.each do |batch| unless batch.empty? notification = { topic: batch.topic, partition: batch.partition, last_offset: batch.last_offset, offset_lag: batch.offset_lag, highwater_mark_offset: batch.highwater_mark_offset, message_count: batch..count, } # Instrument an event immediately so that subscribers don't have to wait until # the block is completed. @instrumenter.instrument("start_process_batch.consumer", notification) @instrumenter.instrument("process_batch.consumer", notification) do begin yield batch @current_offsets[batch.topic][batch.partition] = batch.last_offset rescue => e offset_range = (batch.first_offset..batch.last_offset) location = "#{batch.topic}/#{batch.partition} in offset range #{offset_range}" backtrace = e.backtrace.join("\n") @logger.error "Exception raised when processing #{location} -- #{e.class}: #{e}\n#{backtrace}" raise ProcessingError.new(batch.topic, batch.partition, offset_range) end end (batch..last) if automatically_mark_as_processed # We've successfully processed a batch from the partition, so we can clear # the pause. pause_for(batch.topic, batch.partition).reset! end @offset_manager.commit_offsets_if_necessary @heartbeat.send_if_necessary return if !@running end # We may not have received any messages, but it's still a good idea to # commit offsets if we've processed messages in the last set of batches. # This also ensures the offsets are retained if we haven't read any messages # since the offset retention period has elapsed. @offset_manager.commit_offsets_if_necessary end end |
#each_message(min_bytes: 1, max_bytes: 10485760, max_wait_time: 1, automatically_mark_as_processed: true) {|message| ... } ⇒ nil
Fetches and enumerates the messages in the topics that the consumer group subscribes to.
Each message is yielded to the provided block. If the block returns without raising an exception, the message will be considered successfully processed. At regular intervals the offset of the most recent successfully processed message in each partition will be committed to the Kafka offset store. If the consumer crashes or leaves the group, the group member that is tasked with taking over processing of these partitions will resume at the last committed offsets.
196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 |
# File 'lib/kafka/consumer.rb', line 196 def (min_bytes: 1, max_bytes: 10485760, max_wait_time: 1, automatically_mark_as_processed: true) @fetcher.configure( min_bytes: min_bytes, max_bytes: max_bytes, max_wait_time: max_wait_time, ) consumer_loop do batches = fetch_batches batches.each do |batch| batch..each do || notification = { topic: .topic, partition: .partition, offset: .offset, offset_lag: batch.highwater_mark_offset - .offset - 1, create_time: .create_time, key: .key, value: .value, } # Instrument an event immediately so that subscribers don't have to wait until # the block is completed. @instrumenter.instrument("start_process_message.consumer", notification) @instrumenter.instrument("process_message.consumer", notification) do begin yield @current_offsets[.topic][.partition] = .offset rescue => e location = "#{.topic}/#{.partition} at offset #{.offset}" backtrace = e.backtrace.join("\n") @logger.error "Exception raised when processing #{location} -- #{e.class}: #{e}\n#{backtrace}" raise ProcessingError.new(.topic, .partition, .offset) end end () if automatically_mark_as_processed @offset_manager.commit_offsets_if_necessary @heartbeat.send_if_necessary return if !@running end # We've successfully processed a batch from the partition, so we can clear # the pause. pause_for(batch.topic, batch.partition).reset! end # We may not have received any messages, but it's still a good idea to # commit offsets if we've processed messages in the last set of batches. # This also ensures the offsets are retained if we haven't read any messages # since the offset retention period has elapsed. @offset_manager.commit_offsets_if_necessary end end |
#mark_message_as_processed(message) ⇒ Object
361 362 363 |
# File 'lib/kafka/consumer.rb', line 361 def () @offset_manager.mark_as_processed(.topic, .partition, .offset) end |
#pause(topic, partition, timeout: nil, max_timeout: nil, exponential_backoff: false) ⇒ nil
Pause processing of a specific topic partition.
When a specific message causes the processor code to fail, it can be a good idea to simply pause the partition until the error can be resolved, allowing the rest of the partitions to continue being processed.
If the timeout
argument is passed, the partition will automatically be
resumed when the timeout expires. If exponential_backoff
is enabled, each
subsequent pause will cause the timeout to double until a message from the
partition has been successfully processed.
134 135 136 137 138 139 140 141 142 143 144 |
# File 'lib/kafka/consumer.rb', line 134 def pause(topic, partition, timeout: nil, max_timeout: nil, exponential_backoff: false) if max_timeout && !exponential_backoff raise ArgumentError, "`max_timeout` only makes sense when `exponential_backoff` is enabled" end pause_for(topic, partition).pause!( timeout: timeout, max_timeout: max_timeout, exponential_backoff: exponential_backoff, ) end |
#paused?(topic, partition) ⇒ Boolean
Whether the topic partition is currently paused.
164 165 166 167 |
# File 'lib/kafka/consumer.rb', line 164 def paused?(topic, partition) pause = pause_for(topic, partition) pause.paused? && !pause.expired? end |
#resume(topic, partition) ⇒ nil
Resume processing of a topic partition.
152 153 154 155 156 |
# File 'lib/kafka/consumer.rb', line 152 def resume(topic, partition) pause_for(topic, partition).resume! seek_to_next(topic, partition) end |
#seek(topic, partition, offset) ⇒ nil
Move the consumer's position in a topic partition to the specified offset.
Note that this has to be done prior to calling #each_message or #each_batch and only has an effect if the consumer is assigned the partition. Typically, you will want to do this in every consumer group member in order to make sure that the member that's assigned the partition knows where to start.
353 354 355 |
# File 'lib/kafka/consumer.rb', line 353 def seek(topic, partition, offset) @offset_manager.seek_to(topic, partition, offset) end |
#send_heartbeat_if_necessary ⇒ Object
365 366 367 |
# File 'lib/kafka/consumer.rb', line 365 def send_heartbeat_if_necessary @heartbeat.send_if_necessary end |
#stop ⇒ nil
Stop the consumer.
The consumer will finish any in-progress work and shut down.
110 111 112 113 |
# File 'lib/kafka/consumer.rb', line 110 def stop @running = false @cluster.disconnect end |
#subscribe(topic, default_offset: nil, start_from_beginning: true, max_bytes_per_partition: 1048576) ⇒ nil
Subscribes the consumer to a topic.
Typically you either want to start reading messages from the very
beginning of the topic's partitions or you simply want to wait for new
messages to be written. In the former case, set start_from_beginning
to true (the default); in the latter, set it to false.
95 96 97 98 99 100 101 102 103 |
# File 'lib/kafka/consumer.rb', line 95 def subscribe(topic, default_offset: nil, start_from_beginning: true, max_bytes_per_partition: 1048576) default_offset ||= start_from_beginning ? :earliest : :latest @group.subscribe(topic) @offset_manager.set_default_offset(topic, default_offset) @fetcher.subscribe(topic, max_bytes_per_partition: max_bytes_per_partition) nil end |