Module: Deimos::ActiveRecordConsume::BatchConsumption

Included in:
Deimos::ActiveRecordConsumer
Defined in:
lib/deimos/active_record_consume/batch_consumption.rb

Overview

Methods for consuming batches of messages and saving them to the database in bulk ActiveRecord operations.

Instance Method Summary collapse

Instance Method Details

#consume_batch(payloads, metadata) ⇒ void

This method returns an undefined value.

Handle a batch of Kafka messages. Batches are split into “slices”, which are groups of independent messages that can be processed together in a single database operation. If two messages in a batch have the same key, we cannot process them in the same operation as they would interfere with each other. Thus they are split



26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
 
# File 'lib/deimos/active_record_consume/batch_consumption.rb', line 26

def consume_batch(payloads, metadata)
  messages = payloads.
    zip(metadata[:keys]).
    map { |p, k| Deimos::Message.new(p, nil, key: k) }

  tag = metadata[:topic]
  Deimos.config.tracer.active_span.set_tag('topic', tag)

  Deimos.instrument('ar_consumer.consume_batch', tag) do
    if @compacted || self.class.config[:no_keys]
      update_database(compact_messages(messages))
    else
      uncompacted_update(messages)
    end
  end
end