Class: Karafka::Connection::Proxy

Inherits:
SimpleDelegator
  • Object
show all
Defined in:
lib/karafka/connection/proxy.rb

Overview

Usually it is ok to use the Rdkafka::Consumer directly because we need 1:1 its functionality. There are however cases where we want to have extra recoveries or other handling of errors and settings. This is where this module comes in handy.

We do not want to wrap and delegate all via a proxy object for performance reasons, but we do still want to be able to alter some functionalities. This wrapper helps us do it when it would be needed

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(obj) ⇒ Proxy

Returns a new instance of Proxy.

Parameters:

  • obj (Rdkafka::Consumer, Proxy) —

    rdkafka consumer or consumer wrapped with proxy



38
39
40
41
42
43
44
 
# File 'lib/karafka/connection/proxy.rb', line 38

def initialize(obj)
  super
  # Do not allow for wrapping proxy with a proxy. This will prevent a case where we might
  # wrap an already wrapped object with another proxy level. Simplifies passing consumers
  # and makes it safe to wrap without type checking
  @wrapped = obj.is_a?(self.class) ? obj.wrapped : obj
end

Instance Attribute Details

#wrapped ⇒ Object Also known as: __getobj__

Returns the value of attribute wrapped.



33
34
35
 
# File 'lib/karafka/connection/proxy.rb', line 33

def wrapped
  @wrapped
end

Instance Method Details

#commit_offsets(tpl = nil, async: true) ⇒ Boolean

Note:

We do not consider no_offset as any problem and we allow to commit offsets even when no stored, because with sync commit, it refreshes the ownership state of the consumer in a sync way.

Non thread-safe message committing method

Parameters:

  • tpl (Rdkafka::Consumer::TopicPartitionList, nil) (defaults to: nil) —

    tpl or nil

  • async (Boolean) (defaults to: true) —

    should the commit happen async or sync (async by default)

Returns:

  • (Boolean) —

    true if offset commit worked, false if we've lost the assignment



172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
 
# File 'lib/karafka/connection/proxy.rb', line 172

def commit_offsets(tpl = nil, async: true)
  c_config = proxy_config.commit

  with_broker_errors_retry(
    wait_time: c_config.wait_time / 1_000.to_f,
    max_attempts: c_config.max_attempts
  ) do
    @wrapped.commit(tpl, async)
  end

  true
rescue Rdkafka::RdkafkaError => e
  case e.code
  when :assignment_lost
    return false
  when :unknown_member_id
    return false
  when :illegal_generation
    return false
  when :no_offset
    return true
  when :coordinator_load_in_progress
    sleep(1)
    retry
  end

  raise e
end

#committed(tpl = nil) ⇒ Rdkafka::Consumer::TopicPartitionList

Similar to #query_watermark_offsets.

Parameters:

  • tpl (Rdkafka::Consumer::TopicPartitionList, nil) (defaults to: nil) —

    tpl or nil for full current assignment tpl usage

Returns:

  • (Rdkafka::Consumer::TopicPartitionList) —

    tpl with committed offsets and metadata



135
136
137
138
139
140
141
142
143
144
145
 
# File 'lib/karafka/connection/proxy.rb', line 135

def committed(tpl = nil)
  c_config = proxy_config.committed

  with_broker_errors_retry(
    # required to be in seconds, not ms
    wait_time: c_config.wait_time / 1_000.to_f,
    max_attempts: c_config.max_attempts
  ) do
    @wrapped.committed(tpl, c_config.timeout)
  end
end

#lag(tpl) ⇒ Hash{String => Hash}

Returns hash with topics and their partitions lags.

Parameters:

  • tpl (Rdkafka::Consumer::TopicPartitionList) —

    list of topics and partitions for which we want to get the lag on the defined CG

Returns:

  • (Hash{String => Hash}) —

    hash with topics and their partitions lags



204
205
206
207
208
209
210
211
212
213
214
 
# File 'lib/karafka/connection/proxy.rb', line 204

def lag(tpl)
  l_config = proxy_config.committed

  with_broker_errors_retry(
    # required to be in seconds, not ms
    wait_time: l_config.wait_time / 1_000.to_f,
    max_attempts: l_config.max_attempts
  ) do
    @wrapped.lag(tpl, l_config.timeout)
  end
end

#metadata(topic_name = nil) ⇒ Rdkafka::Metadata

Returns rdkafka metadata object with the requested details.

Parameters:

  • topic_name (String, nil) (defaults to: nil) —

    Name of the topic we're interested in or nil if we want to get info on all topics

Returns:

  • (Rdkafka::Metadata) —

    rdkafka metadata object with the requested details



219
220
221
222
223
224
225
226
227
228
229
 
# File 'lib/karafka/connection/proxy.rb', line 219

def metadata(topic_name = nil)
  m_config = proxy_config.metadata

  with_broker_errors_retry(
    # required to be in seconds, not ms
    wait_time: m_config.wait_time / 1_000.to_f,
    max_attempts: m_config.max_attempts
  ) do
    @wrapped.metadata(topic_name, m_config.timeout)
  end
end

#offsets_for_times(tpl) ⇒ Rdkafka::Consumer::TopicPartitionList

Similar to #query_watermark_offsets, this method can be sensitive to latency. We handle this the same way

Parameters:

  • tpl (Rdkafka::Consumer::TopicPartitionList) —

    tpl to get time offsets

Returns:

  • (Rdkafka::Consumer::TopicPartitionList) —

    tpl with time offsets



118
119
120
121
122
123
124
125
126
127
128
 
# File 'lib/karafka/connection/proxy.rb', line 118

def offsets_for_times(tpl)
  l_config = proxy_config.offsets_for_times

  with_broker_errors_retry(
    # required to be in seconds, not ms
    wait_time: l_config.wait_time / 1_000.to_f,
    max_attempts: l_config.max_attempts
  ) do
    @wrapped.offsets_for_times(tpl, l_config.timeout)
  end
end

#query_watermark_offsets(topic, partition) ⇒ Array<Integer, Integer>

Proxies the #query_watermark_offsets with extra recovery from timeout problems. We impose our own custom timeout to make sure, that high-latency clusters and overloaded clusters can handle our requests.

Parameters:

  • topic (String) —

    topic name

  • partition (Integer) —

    partition number

Returns:

  • (Array<Integer, Integer>) —

    watermark offsets



53
54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/karafka/connection/proxy.rb', line 53

def query_watermark_offsets(topic, partition)
  l_config = proxy_config.query_watermark_offsets

  # For newly created topics or in cases where we're trying to get them but there is no
  # leader, this can fail. It happens more often for new topics under KRaft, however we
  # still want to make sure things operate as expected even then
  with_broker_errors_retry(
    # required to be in seconds, not ms
    wait_time: l_config.wait_time / 1_000.to_f,
    max_attempts: l_config.max_attempts
  ) do
    @wrapped.query_watermark_offsets(topic, partition, l_config.timeout)
  end
end

#read_partition_offsets(topic_partition_offsets, isolation_level: nil) ⇒ Array<Hash>

Note:

The ListOffsets admin API resolves :latest to the high watermark even with isolation_level set to read_committed, unlike #query_watermark_offsets, which returns the last stable offset for a read_committed consumer. On a topic with an in-flight transaction :latest therefore includes the uncommitted messages. Non-transactional topics are unaffected.

Note:

Specs given as a literal hash need explicit curly braces: a brace-less trailing hash is parsed by Ruby as keyword arguments, which this method (having the isolation_level: keyword) would reject as a missing positional argument

Resolves offsets of many partitions with one batched ListOffsets request instead of one roundtrip per partition. This is the same broker query as #query_watermark_offsets, only batched, so it is sensitive to the same latencies and leader-election hiccups and we recover from them the same way and with the same settings.

Named after Admin::Topics#read_partition_offsets (same specs in, same offsets out) and not after the underlying #list_offsets, so we do not shadow the rdkafka method of that name: it returns a handle to wait on, while we return the already resolved offsets.

Parameters:

  • topic_partition_offsets (Hash{String => Array<Hash>}) —

    topics with arrays of partition offset specs, each with a :partition and an :offset (:earliest, :latest, :max_timestamp or an integer timestamp in ms)

  • isolation_level (Integer, nil) (defaults to: nil) —

    isolation level with which the offsets should be resolved or nil for the librdkafka default (read_uncommitted)

Returns:

  • (Array<Hash>) —

    resolved offsets, each with :topic, :partition and :offset



91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
 
# File 'lib/karafka/connection/proxy.rb', line 91

def read_partition_offsets(topic_partition_offsets, isolation_level: nil)
  l_config = proxy_config.query_watermark_offsets

  with_broker_errors_retry(
    # required to be in seconds, not ms
    wait_time: l_config.wait_time / 1_000.to_f,
    max_attempts: l_config.max_attempts
  ) do
    @wrapped
      .list_offsets(topic_partition_offsets, isolation_level: isolation_level)
      .wait(max_wait_timeout_ms: l_config.timeout)
      .offsets
  rescue Rdkafka::AbstractHandle::WaitTimeoutError
    # Align with the error contract of the per-partition watermark query, so a slow broker
    # is retried here instead of escaping as a handle wait timeout
    raise Rdkafka::RdkafkaError.new(
      Rdkafka::Bindings::RD_KAFKA_RESP_ERR__TIMED_OUT,
      "Error querying offsets of '#{topic_partition_offsets.keys.join(", ")}'"
    )
  end
end

#store_offset(message, metadata = nil) ⇒ Boolean

When we cannot store an offset, it means we no longer own the partition

Non thread-safe offset storing method

Parameters:

Returns:

  • (Boolean) —

    true if we could store the offset (if we still own the partition)



153
154
155
156
157
158
159
160
161
162
163
 
# File 'lib/karafka/connection/proxy.rb', line 153

def store_offset(message, metadata = nil)
  @wrapped.store_offset(message, metadata)

  true
rescue Rdkafka::RdkafkaError => e
  return false if e.code == :assignment_lost
  return false if e.code == :state
  return false if e.code == :illegal_generation

  raise e
end