Class: Karafka::BaseConsumer

Inherits:
Object
  • Object
show all
Extended by:
Forwardable
Includes:
Core::Taggable
Defined in:
lib/karafka/base_consumer.rb

Overview

Base consumer from which all Karafka consumers should inherit

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ BaseConsumer

Creates new consumer and assigns it an id



35
36
37
38
 
# File 'lib/karafka/base_consumer.rb', line 35

def initialize
  @id = SecureRandom.hex(6)
  @used = false
end

Instance Attribute Details

#client ⇒ Karafka::Connection::Client

Returns kafka connection client.

Returns:



28
29
30
 
# File 'lib/karafka/base_consumer.rb', line 28

def client
  @client
end

#coordinator ⇒ Karafka::Processing::ConsumerGroups::Coordinator

Returns coordinator.

Returns:



30
31
32
 
# File 'lib/karafka/base_consumer.rb', line 30

def coordinator
  @coordinator
end

#id ⇒ String (readonly)

Returns id of the current consumer.

Returns:

  • (String) —

    id of the current consumer



24
25
26
 
# File 'lib/karafka/base_consumer.rb', line 24

def id
  @id
end

#messages ⇒ Karafka::Messages::Messages

Returns current messages batch.

Returns:



26
27
28
 
# File 'lib/karafka/base_consumer.rb', line 26

def messages
  @messages
end

#producer ⇒ Waterdrop::Producer

Returns producer instance.

Returns:

  • (Waterdrop::Producer) —

    producer instance



32
33
34
 
# File 'lib/karafka/base_consumer.rb', line 32

def producer
  @producer
end

Instance Method Details

#inspect ⇒ String

Returns a string representation of the consumer instance for debugging purposes.

This method provides a safe inspection that avoids walking through potentially large nested objects like messages, client connections, or coordinator state that could cause performance issues during logging or debugging.

Returns:

  • (String) —

    formatted string containing essential consumer information including consumer ID, topic name, partition number, usage status, message count, and revocation status



236
237
238
239
240
241
242
243
244
245
246
247
 
# File 'lib/karafka/base_consumer.rb', line 236

def inspect
  parts = [
    "id=#{@id}",
    "topic=#{topic&.name.inspect}",
    "partition=#{partition}",
    "used=#{@used}",
    "messages_count=#{@messages&.count}",
    "revoked=#{coordinator&.revoked?}"
  ]

  "#<#{self.class.name}:#{format("%#x", object_id)} #{parts.join(" ")}>"
end

#on_after_consume ⇒ Object

Note:

This should not be used by the end users as it is part of the lifecycle of things but not as part of the public api.

Note:

We handle and report errors here because of flows that could fail. For example a DLQ flow could fail if it was not able to dispatch the DLQ message. Other "non-user" based flows do not interact with external systems and their errors are expected to bubble up



135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
 
# File 'lib/karafka/base_consumer.rb', line 135

def on_after_consume
  handle_after_consume
# Same containment rationale as in `#on_consume`: an error escaping this method would skip
# the retry below and the next successful batch would auto-mark its offsets, committing
# past the failed batch. The after-consume flow runs user-extensible code as well (DLQ
# strategies, dispatch enhancements), so it needs the same class-agnostic protection
rescue Exception => e
  monitor.instrument(
    "error.occurred",
    error: e,
    caller: self,
    seek_offset: seek_offset,
    type: "consumer.after_consume.error"
  )

  retry_after_pause
end

#on_before_consume ⇒ Object

Note:

This should not be used by the end users as it is part of the lifecycle of things and not as part of the public api. This can act as a hook when creating non-blocking consumers and doing other advanced stuff

Can be used to run preparation code in the worker



71
72
73
74
75
76
77
78
 
# File 'lib/karafka/base_consumer.rb', line 71

def on_before_consume
  messages.metadata.processed_at = Time.now
  messages.metadata.freeze

  # We run this after the full metadata setup, so we can use all the messages information
  # if needed
  handle_before_consume
end

#on_before_schedule_consume ⇒ Object

Note:

This should not be used by the end users as it is part of the lifecycle of things and not as a part of the public api. This should not perform any extensive operations as it is blocking and running in the listener thread.

Can be used to run preparation code prior to the job being enqueued



60
61
62
63
 
# File 'lib/karafka/base_consumer.rb', line 60

def on_before_schedule_consume
  @used = true
  handle_before_schedule_consume
end

#on_before_schedule_eofed ⇒ Object

Can be used to run code prior to scheduling of eofed execution



154
155
156
 
# File 'lib/karafka/base_consumer.rb', line 154

def on_before_schedule_eofed
  handle_before_schedule_eofed
end

#on_before_schedule_idle ⇒ Object

Can be used to run code prior to scheduling of idle execution



174
175
176
 
# File 'lib/karafka/base_consumer.rb', line 174

def on_before_schedule_idle
  handle_before_schedule_idle
end

#on_before_schedule_revoked ⇒ Object

Can be used to run code prior to scheduling of revoked execution



188
189
190
 
# File 'lib/karafka/base_consumer.rb', line 188

def on_before_schedule_revoked
  handle_before_schedule_revoked
end

#on_before_schedule_shutdown ⇒ Object

Can be used to run code prior to scheduling of revoked execution



209
210
211
 
# File 'lib/karafka/base_consumer.rb', line 209

def on_before_schedule_shutdown
  handle_before_schedule_shutdown
end

#on_consume ⇒ Boolean

Note:

We keep the seek offset tracking, and use it to compensate for async offset flushing that may not yet kick in when error occurs. That way we pause always on the last processed message.

Executes the default consumer flow.

Returns:

  • (Boolean) —

    true if there was no exception, otherwise false.



105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
 
# File 'lib/karafka/base_consumer.rb', line 105

def on_consume
  handle_consume
# Containment is intentionally broader than StandardError: any error escaping this method
# would bypass `#on_after_consume` in the worker, skipping the retry/pause flow entirely.
# The next successful batch would then auto-mark its offsets, durably committing past the
# failed batch - a silent at-least-once violation. Errors like SystemStackError or the
# ScriptError family (e.g. LoadError surfacing from a Rails autoload hiccup) are per-message
# failures and go through the regular retry flow like any other processing error.
# Process-critical errors additionally trigger a graceful shutdown via the auto-subscribed
# `Instrumentation::CriticalErrorsListener` watching this very instrumentation - the
# recorded failure plus the retry pause keep this partition protected during the shutdown
# window and the batch is redelivered after restart.
rescue Exception => e
  monitor.instrument(
    "error.occurred",
    error: e,
    caller: self,
    seek_offset: seek_offset,
    type: "consumer.consume.error"
  )
end

#on_eofed ⇒ Object

Trigger method for running on eof without messages



159
160
161
162
163
164
165
166
167
168
169
 
# File 'lib/karafka/base_consumer.rb', line 159

def on_eofed
  handle_eofed
rescue => e
  monitor.instrument(
    "error.occurred",
    error: e,
    caller: self,
    seek_offset: seek_offset,
    type: "consumer.eofed.error"
  )
end

#on_idle ⇒ Object

Trigger method for running on idle runs without messages



181
182
183
 
# File 'lib/karafka/base_consumer.rb', line 181

def on_idle
  handle_idle
end

#on_initialized ⇒ Object

Trigger method running after consumer is fully initialized.



43
44
45
46
47
48
49
50
51
52
 
# File 'lib/karafka/base_consumer.rb', line 43

def on_initialized
  handle_initialized
rescue => e
  monitor.instrument(
    "error.occurred",
    error: e,
    caller: self,
    type: "consumer.initialized.error"
  )
end

#on_revoked ⇒ Object

Trigger method for running on partition revocation.



195
196
197
198
199
200
201
202
203
204
 
# File 'lib/karafka/base_consumer.rb', line 195

def on_revoked
  handle_revoked
rescue => e
  monitor.instrument(
    "error.occurred",
    error: e,
    caller: self,
    type: "consumer.revoked.error"
  )
end

#on_shutdown ⇒ Object

Trigger method for running on shutdown.



216
217
218
219
220
221
222
223
224
225
 
# File 'lib/karafka/base_consumer.rb', line 216

def on_shutdown
  handle_shutdown
rescue => e
  monitor.instrument(
    "error.occurred",
    error: e,
    caller: self,
    type: "consumer.shutdown.error"
  )
end

#on_wrap(action) ⇒ Object

Executes the default wrapping flow

Parameters:

  • action (Symbol) 


85
86
87
88
89
90
91
92
93
94
 
# File 'lib/karafka/base_consumer.rb', line 85

def on_wrap(action, &)
  handle_wrap(action, &)
rescue => e
  monitor.instrument(
    "error.occurred",
    error: e,
    caller: self,
    type: "consumer.wrap.error"
  )
end