Class: Karafka::Processing::ConsumerGroups::Jobs::Consume

Inherits:
Jobs::Base
  • Object
show all
Defined in:
lib/karafka/processing/consumer_groups/jobs/consume.rb

Overview

The main job type. It runs the executor that triggers given topic partition messages processing in an underlying consumer instance.

Instance Attribute Summary collapse

Attributes inherited from Jobs::Base

#executor

Instance Method Summary collapse

Methods inherited from Jobs::Base

#finish!, #finished?, #non_blocking?, #wrap

Constructor Details

#initialize(executor, messages) ⇒ Consume

Parameters:



19
20
21
22
23
 
# File 'lib/karafka/processing/consumer_groups/jobs/consume.rb', line 19

def initialize(executor, messages)
  @executor = executor
  @messages = messages
  super()
end

Instance Attribute Details

#messages ⇒ Array<Rdkafka::Consumer::Message> (readonly)

Returns array with messages.

Returns:

  • (Array<Rdkafka::Consumer::Message>) —

    array with messages



11
12
13
 
# File 'lib/karafka/processing/consumer_groups/jobs/consume.rb', line 11

def messages
  @messages
end

Instance Method Details

#after_call ⇒ Object

Runs any error handling and other post-consumption stuff on the executor



42
43
44
 
# File 'lib/karafka/processing/consumer_groups/jobs/consume.rb', line 42

def after_call
  executor.after_consume
end

#before_call ⇒ Object

Runs the before consumption preparations on the executor



32
33
34
 
# File 'lib/karafka/processing/consumer_groups/jobs/consume.rb', line 32

def before_call
  executor.before_consume
end

#before_schedule ⇒ Object

Runs all the preparation code on the executor that needs to happen before the job is scheduled.



27
28
29
 
# File 'lib/karafka/processing/consumer_groups/jobs/consume.rb', line 27

def before_schedule
  executor.before_schedule_consume(@messages)
end

#call ⇒ Object

Runs the given executor



37
38
39
 
# File 'lib/karafka/processing/consumer_groups/jobs/consume.rb', line 37

def call
  executor.consume
end