Class: Karafka::Processing::Executor
- Inherits:
-
Object
- Object
- Karafka::Processing::Executor
- Extended by:
- Forwardable
- Defined in:
- lib/karafka/processing/executor.rb
Overview
Executors are not removed after partition is revoked. They are not that big and will be re-used in case of a re-claim
Executors:
-
run consumers code (for ‘#call`) or run given preparation / teardown operations when needed from separate threads.
-
they re-create consumer instances in case of partitions that were revoked and assigned back.
Instance Attribute Summary collapse
-
#coordinator ⇒ Karafka::Processing::Coordinator
readonly
Coordinator for this executor.
-
#group_id ⇒ String
readonly
Subscription group id to which a given executor belongs.
-
#id ⇒ String
readonly
Unique id that we use to ensure, that we use for state tracking.
-
#messages ⇒ Karafka::Messages::Messages
readonly
Messages batch.
Instance Method Summary collapse
-
#after_consume ⇒ Object
Runs consumer after consumption code.
-
#before_consume ⇒ Object
Runs setup and warm-up code in the worker prior to running the consumption.
-
#before_enqueue(messages) ⇒ Object
Allows us to prepare the consumer in the listener thread prior to the job being send to the queue.
-
#consume ⇒ Object
Runs consumer data processing against given batch and handles failures and errors.
-
#idle ⇒ Object
Runs consumer idle operations This may include house-keeping or other state management changes that can occur but that not mean there are any new messages available for the end user to process.
-
#initialize(group_id, client, coordinator) ⇒ Executor
constructor
A new instance of Executor.
-
#revoked ⇒ Object
Runs the controller ‘#revoked` method that should be triggered when a given consumer is no longer needed due to partitions reassignment.
-
#shutdown ⇒ Object
Runs the controller ‘#shutdown` method that should be triggered when a given consumer is no longer needed as we’re closing the process.
Constructor Details
#initialize(group_id, client, coordinator) ⇒ Executor
Returns a new instance of Executor.
34 35 36 37 38 39 |
# File 'lib/karafka/processing/executor.rb', line 34 def initialize(group_id, client, coordinator) @id = SecureRandom.hex(6) @group_id = group_id @client = client @coordinator = coordinator end |
Instance Attribute Details
#coordinator ⇒ Karafka::Processing::Coordinator (readonly)
Returns coordinator for this executor.
29 30 31 |
# File 'lib/karafka/processing/executor.rb', line 29 def coordinator @coordinator end |
#group_id ⇒ String (readonly)
Returns subscription group id to which a given executor belongs.
23 24 25 |
# File 'lib/karafka/processing/executor.rb', line 23 def group_id @group_id end |
#id ⇒ String (readonly)
Returns unique id that we use to ensure, that we use for state tracking.
20 21 22 |
# File 'lib/karafka/processing/executor.rb', line 20 def id @id end |
#messages ⇒ Karafka::Messages::Messages (readonly)
Returns messages batch.
26 27 28 |
# File 'lib/karafka/processing/executor.rb', line 26 def @messages end |
Instance Method Details
#after_consume ⇒ Object
Runs consumer after consumption code
78 79 80 |
# File 'lib/karafka/processing/executor.rb', line 78 def after_consume consumer.on_after_consume end |
#before_consume ⇒ Object
Runs setup and warm-up code in the worker prior to running the consumption
67 68 69 |
# File 'lib/karafka/processing/executor.rb', line 67 def before_consume consumer.on_before_consume end |
#before_enqueue(messages) ⇒ Object
Allows us to prepare the consumer in the listener thread prior to the job being send to the queue. It also allows to run some code that is time sensitive and cannot wait in the queue as it could cause starvation.
46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 |
# File 'lib/karafka/processing/executor.rb', line 46 def before_enqueue() # Recreate consumer with each batch if persistence is not enabled # We reload the consumers with each batch instead of relying on some external signals # when needed for consistency. That way devs may have it on or off and not in this # middle state, where re-creation of a consumer instance would occur only sometimes @consumer = nil unless topic.consumer_persistence # First we build messages batch... consumer. = Messages::Builders::Messages.call( , topic, partition, # the moment we've received the batch or actually the moment we've enqueued it, # but good enough Time.now ) consumer.on_before_enqueue end |
#consume ⇒ Object
Runs consumer data processing against given batch and handles failures and errors.
72 73 74 75 |
# File 'lib/karafka/processing/executor.rb', line 72 def consume # We run the consumer client logic... consumer.on_consume end |
#idle ⇒ Object
Runs consumer idle operations This may include house-keeping or other state management changes that can occur but that not mean there are any new messages available for the end user to process
85 86 87 88 89 90 91 92 93 94 95 96 97 |
# File 'lib/karafka/processing/executor.rb', line 85 def idle # Initializes the messages set in case idle operation would happen before any processing # This prevents us from having no messages object at all as the messages object and # its metadata may be used for statistics consumer. ||= Messages::Builders::Messages.call( [], topic, partition, Time.now ) consumer.on_idle end |
#revoked ⇒ Object
Clearing the consumer will ensure, that if we get the partition back, it will be handled with a consumer with a clean state.
We run it only when consumer was present, because presence indicates, that at least a single message has been consumed.
We do not reset the consumer but we indicate need for recreation instead, because after the revocation, there still may be ‘#after_consume` running that needs a given consumer instance.
Runs the controller ‘#revoked` method that should be triggered when a given consumer is no longer needed due to partitions reassignment.
111 112 113 |
# File 'lib/karafka/processing/executor.rb', line 111 def revoked consumer.on_revoked if @consumer end |
#shutdown ⇒ Object
While we do not need to clear the consumer here, it’s a good habit to clean after work is done.
Runs the controller ‘#shutdown` method that should be triggered when a given consumer is no longer needed as we’re closing the process.
120 121 122 123 124 |
# File 'lib/karafka/processing/executor.rb', line 120 def shutdown # There is a case, where the consumer no longer exists because it was revoked, in case like # that we do not build a new instance and shutdown should not be triggered. consumer.on_shutdown if @consumer end |