Class: Tina4::QueueBackends::KafkaBackend

Inherits:
Object
  • Object
show all
Defined in:
lib/tina4/queue_backends/kafka_backend.rb

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(options = {}) ⇒ KafkaBackend

Returns a new instance of KafkaBackend.



6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
 
# File 'lib/tina4/queue_backends/kafka_backend.rb', line 6

def initialize(options = {})
  require "rdkafka"
  @brokers = options[:brokers] || "localhost:9092"
  @group_id = options[:group_id] || "tina4_consumer_group"
  @max_retries = options[:max_retries] || 3

  security = self.class._security_config
  # Kept so dead_letters() can build a short-lived READER consumer with
  # its own group id, which reads the dead-letter topic from the start
  # without disturbing the main subscription or committing offsets.
  @security = security

  producer_config = {
    "bootstrap.servers" => @brokers
  }.merge(security)
  @producer = Rdkafka::Config.new(producer_config).producer

  consumer_config = {
    "bootstrap.servers" => @brokers,
    "group.id" => @group_id,
    "auto.offset.reset" => "earliest",
    "enable.auto.commit" => "false"
  }.merge(security)
  @consumer = Rdkafka::Config.new(consumer_config).consumer
  @subscribed_topics = []
rescue LoadError
  raise "Kafka backend requires the 'rdkafka' gem. Install with: gem install rdkafka"
end

Instance Attribute Details

#max_retries ⇒ Object

Build SSL/SASL client config from env (for a TLS broker/proxy).

Mirrors tina4_python KafkaConnector._security_config: each setting is read from the Tina4-namespaced env var first (TINA4_KAFKA_) and falls back to the bare librdkafka-convention name (KAFKA_) that many Kafka deployments already set. Honours security.protocol (e.g. SSL, SASL_SSL), ssl.ca.location, and optional SASL (mechanism / username / password). Unset keys are omitted, leaving librdkafka's PLAINTEXT default. Queue propagates its own configuration onto the backend after construction.



44
45
46
 
# File 'lib/tina4/queue_backends/kafka_backend.rb', line 44

def max_retries
  @max_retries
end

Class Method Details

._security_config ⇒ Object 



46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/tina4/queue_backends/kafka_backend.rb', line 46

def self._security_config
  # rdkafka key -> env suffix (read as TINA4_KAFKA_<suffix>, then KAFKA_<suffix>)
  mapping = {
    "security.protocol" => "SECURITY_PROTOCOL",
    "ssl.ca.location" => "SSL_CA_LOCATION",
    "sasl.mechanism" => "SASL_MECHANISM",
    "sasl.username" => "SASL_USERNAME",
    "sasl.password" => "SASL_PASSWORD"
  }
  config = {}
  mapping.each do |rdk, suffix|
    value = env_value("TINA4_KAFKA_#{suffix}") || env_value("KAFKA_#{suffix}")
    config[rdk] = value if value
  end
  config
end

Instance Method Details

#acknowledge(_message) ⇒ Object 



151
152
153
 
# File 'lib/tina4/queue_backends/kafka_backend.rb', line 151

def acknowledge(_message)
  @consumer.commit if @last_message
end

#clear(_topic) ⇒ Object

Raises:

  • (NotImplementedError) 


298
299
300
301
302
303
304
305
 
# File 'lib/tina4/queue_backends/kafka_backend.rb', line 298

def clear(_topic)
  raise NotImplementedError,
        "The kafka queue backend cannot perform clear(): a log cannot " \
        "delete records on demand - retention is time- and size-based and " \
        "set on the topic. Returning 0 would claim the queue was emptied " \
        "when it was not. Configure topic retention, or use the file or " \
        "mongodb backend."
end

#close ⇒ Object

Close the rdkafka producer and consumer (leaving the consumer group).

IDEMPOTENT by construction: the handles are dropped in an ensure, so a second close finds nothing and returns. Before 3.13.95 they were left set, and rdkafka raises on closing an already-closed consumer - so a shutdown path that ran twice crashed on the second pass.



313
314
315
316
317
318
319
320
 
# File 'lib/tina4/queue_backends/kafka_backend.rb', line 313

def close
  @producer&.close
  @consumer&.close
ensure
  @producer = nil
  @consumer = nil
  @subscribed_topics = []
end

#complete(message) ⇒ Object

Terminal ack. Job#complete calls backend.complete, and this backend only had acknowledge() - so job.complete was a silent no-op on kafka and the consumer-group offset was never committed, making the broker redeliver every completed record.



177
178
179
 
# File 'lib/tina4/queue_backends/kafka_backend.rb', line 177

def complete(message)
  acknowledge(message)
end

#dead_letter(message) ⇒ Object 



159
160
161
162
163
164
165
166
167
168
169
170
171
 
# File 'lib/tina4/queue_backends/kafka_backend.rb', line 159

def dead_letter(message)
  # attempts and error MUST ride along, else every Kafka dead letter reads
  # back as attempts=0 with no reason while file/mongodb report the real
  # values.
  dead_msg = Tina4::Job.new(
    topic: "#{message.topic}.dead_letter",
    payload: message.payload,
    id: message.id,
    attempts: message.attempts,
    error: message.error
  )
  enqueue(dead_msg)
end

#dead_letters(topic, max_retries: 3) ⇒ Object

Dead-lettered jobs, read back from the .dead_letter topic this backend produces to itself. Kafka's log is not queryable by job state, but it IS readable from the beginning - so this ANSWERS rather than refusing, and a dead-letter handler written against the file backend finds the same jobs here (invariant 3).

Uses a short-lived consumer with a UNIQUE group id and never commits, so it always reads from the start and a read never consumes. Reading with the main consumer would move its offsets and steal the subscription.



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
255
256
257
258
259
260
 
# File 'lib/tina4/queue_backends/kafka_backend.rb', line 230

def dead_letters(topic, max_retries: 3)
  name = "#{topic}.dead_letter"
  reader = Rdkafka::Config.new({
    "bootstrap.servers" => @brokers,
    "group.id" => "#{@group_id}.reader.#{Process.pid}.#{object_id}",
    "auto.offset.reset" => "earliest",
    "enable.auto.commit" => "false"
  }.merge(@security)).consumer
  out = []
  begin
    reader.subscribe(name)
    deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) +
               (ENV["TINA4_KAFKA_ASSIGN_TIMEOUT"] || "15").to_f
    loop do
      msg = reader.poll(500)
      if msg
        data = JSON.parse(msg.payload)
        data["status"] = "dead"
        out << data
        next
      end
      break if Process.clock_gettime(Process::CLOCK_MONOTONIC) >= deadline
    end
  rescue Rdkafka::RdkafkaError
    # An absent topic means nothing has been dead-lettered yet.
    nil
  ensure
    reader.close
  end
  out
end

#dequeue(topic) ⇒ Object 



97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
 
# File 'lib/tina4/queue_backends/kafka_backend.rb', line 97

def dequeue(topic)
  first = !@subscribed_topics.include?(topic)
  if first
    @consumer.subscribe(topic)
    @subscribed_topics << topic
  end

  # The first poll after subscribing must drive the consumer-group join +
  # partition assignment, which takes several seconds on a cold broker.
  # Until partitions are assigned, poll returns nil even when the topic
  # already has messages -- so a single poll made dequeue return nil right
  # after enqueue. Poll in a bounded loop on first subscribe (deadline
  # TINA4_KAFKA_ASSIGN_TIMEOUT, default 15s); steady state stays one ~1s poll.
  deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) +
             (first ? (ENV["TINA4_KAFKA_ASSIGN_TIMEOUT"] || "15").to_f : 1.0)
  msg = nil
  loop do
    candidate = @consumer.poll(500)
    if candidate
      msg = candidate
      break
    end
    break if Process.clock_gettime(Process::CLOCK_MONOTONIC) >= deadline
  end
  return nil unless msg

  data = JSON.parse(msg.payload)
  @last_message = msg

  # attempts and error MUST be carried back. Rebuilding from
  # topic/payload/id alone reset attempts to 0 on every redelivery, so
  # fail()'s attempts >= max_retries check could never trip and a poison
  # record would be re-produced forever instead of dead-lettering.
  Tina4::Job.new(
    topic: data["topic"],
    payload: data["payload"],
    id: data["id"],
    attempts: data["attempts"] || 0,
    error: data["error"]
  )
rescue Rdkafka::RdkafkaError
  nil
end

#enqueue(message) ⇒ Object 



73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/tina4/queue_backends/kafka_backend.rb', line 73

def enqueue(message)
  if message.priority.to_i > 0
    raise NotImplementedError,
          "The kafka queue backend cannot honour push(priority): Kafka has no " \
          "priority concept at all - a consumer reads a partition in offset " \
          "order. Use the file or mongodb backend for prioritised jobs."
  end

  if message.available_at
    raise NotImplementedError,
          "The kafka queue backend cannot honour push(delay_seconds): Kafka " \
          "has no per-message delay at all. A consumer reads a partition in " \
          "offset order, so a delayed record would stall every record behind " \
          "it. Use the file or mongodb backend for delayed jobs, or schedule " \
          "the push itself."
  end

  @producer.produce(
    topic: message.topic,
    payload: message.to_json,
    key: message.id
  ).wait
end

#fail(job, error = "") ⇒ Object

Record a failed attempt, then retry it or dead-letter it.

This did not exist. Job#fail guarded on respond_to?(:fail) and silently degraded to in-memory bookkeeping, so job.fail() NEVER reached Kafka: the offset was never committed and no dead letter was produced.

A retry RE-PRODUCES a record carrying the new count, because a Kafka record is immutable and carries no delivery counter. The offset is committed LAST so a crash in between redelivers rather than loses - at-least-once, which is the contract.



191
192
193
194
195
196
197
198
199
200
 
# File 'lib/tina4/queue_backends/kafka_backend.rb', line 191

def fail(job, error = "")
  job.attempts += 1
  job.error = error
  if job.attempts >= @max_retries
    dead_letter(job)
  else
    enqueue(job)
  end
  acknowledge(job)
end

#failed(_topic, max_retries: 3) ⇒ Object

Raises:

  • (NotImplementedError) 


262
263
264
265
266
267
268
269
270
 
# File 'lib/tina4/queue_backends/kafka_backend.rb', line 262

def failed(_topic, max_retries: 3)
  raise NotImplementedError,
        "The kafka queue backend cannot answer failed(): a job that failed " \
        "but is still retryable is re-produced to the main topic, and a log " \
        "cannot be queried by job state, so it cannot be told apart from a " \
        "normal pending record. Returning an empty list would claim nothing " \
        "has failed. Use dead_letters() for exhausted jobs, or the file or " \
        "mongodb backend to enumerate retryable failures."
end

#purge(_topic, _status) ⇒ Object

Raises:

  • (NotImplementedError) 


290
291
292
293
294
295
296
 
# File 'lib/tina4/queue_backends/kafka_backend.rb', line 290

def purge(_topic, _status)
  raise NotImplementedError,
        "The kafka queue backend cannot perform purge(): a log cannot " \
        "delete records by status - retention is time- and size-based and " \
        "set on the topic. Returning 0 would claim nothing needed purging. " \
        "Configure topic retention, or use the file or mongodb backend."
end

#requeue(message) ⇒ Object 



155
156
157
 
# File 'lib/tina4/queue_backends/kafka_backend.rb', line 155

def requeue(message)
  enqueue(message)
end

#retry(job, delay_seconds: 0) ⇒ Object 



202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
 
# File 'lib/tina4/queue_backends/kafka_backend.rb', line 202

def retry(job, delay_seconds: 0)
  if delay_seconds.to_f > 0
    raise NotImplementedError,
          "The kafka queue backend cannot honour retry(delay_seconds): " \
          "Kafka has no per-message delay, for the same reason " \
          "push(delay_seconds) is refused - a consumer reads a partition " \
          "in offset order, so a delayed record stalls every record behind " \
          "it. Re-producing immediately while silently dropping the delay " \
          "would run the job far sooner than asked. Use the file or mongodb " \
          "backend for delayed retries."
  end

  job.attempts += 1
  job.error = nil
  enqueue(job)
  acknowledge(job)
  true
end

#retry_failed(_topic, max_retries: 3) ⇒ Object

Raises:

  • (NotImplementedError) 


272
273
274
275
276
277
278
279
 
# File 'lib/tina4/queue_backends/kafka_backend.rb', line 272

def retry_failed(_topic, max_retries: 3)
  raise NotImplementedError,
        "The kafka queue backend cannot perform retry_failed(): it must " \
        "first enumerate the failed-but-retryable jobs, which a log cannot " \
        "be queried for. Returning 0 would claim nothing needed retrying. " \
        "Use retry(job_id) with an id you already hold, or the file or " \
        "mongodb backend."
end

#retry_job(_topic, job_id: nil, delay_seconds: 0) ⇒ Object

Raises:

  • (NotImplementedError) 


281
282
283
284
285
286
287
288
 
# File 'lib/tina4/queue_backends/kafka_backend.rb', line 281

def retry_job(_topic, job_id: nil, delay_seconds: 0)
  raise NotImplementedError,
        "The kafka queue backend cannot perform retry(job_id): a log cannot " \
        "retract a record, so the dead letter would stay on the " \
        ".dead_letter topic and be revived again on the next call, " \
        "duplicating the job. Re-produce it yourself with push() if that " \
        "is what you want, or use the file or mongodb backend."
end

#size(_topic) ⇒ Object

Kafka has no queue depth: a log is a sequence of offsets, not a queue, and computing a "remaining" count means an admin round-trip per call. ADR-0022 decision 5 records 0 as the documented answer, which Python and PHP already return. Ruby had no size method at all, so queue.size raised NoMethodError on kafka - a method that does not resolve at all, which is what invariant 1 forbids.



147
148
149
 
# File 'lib/tina4/queue_backends/kafka_backend.rb', line 147

def size(_topic)
  0
end