Class: Karafka::Connection::PausesManager

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

Overview

Partitions pauses management abstraction layer. It aggregates all the pauses for all the partitions that we're working with.

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Karafka::Connection::PausesManager

Returns pauses manager.



9
10
11
12
13
 
# File 'lib/karafka/connection/pauses_manager.rb', line 9

def initialize
  @pauses = Hash.new do |h, k|
    h[k] = {}
  end
end

Instance Method Details

#fetch(topic, partition) ⇒ Karafka::TimeTrackers::Pause

Creates or fetches pause tracker of a given topic partition.

Parameters:

Returns:



20
21
22
23
24
25
26
 
# File 'lib/karafka/connection/pauses_manager.rb', line 20

def fetch(topic, partition)
  @pauses[topic][partition] ||= TimeTrackers::Pause.new(
    timeout: topic.pause_timeout,
    max_timeout: topic.pause_max_timeout,
    exponential_backoff: topic.pause_with_exponential_backoff
  )
end

#resume {|topic, partition| ... } ⇒ Object

Resumes processing of partitions for which pause time has ended.

Yield Parameters:



49
50
51
52
53
54
55
56
57
58
59
60
 
# File 'lib/karafka/connection/pauses_manager.rb', line 49

def resume
  @pauses.each do |topic, partitions|
    partitions.each do |partition, pause|
      next unless pause.paused?
      next unless pause.expired?

      pause.resume

      yield(topic, partition)
    end
  end
end

#revoke(topic, partition) ⇒ Object

Resets the attempt count of a given topic partition pause tracker.

Used on revocation so that a later reclaim of the same partition starts counting retry attempts from zero instead of carrying the stale count across the rebalance. We reset rather than remove the tracker because the pause itself may still need to be resumed after the reclaim (the partition can be re-paused via the retained paused offsets on rebalance).

A coordinator and its pause tracker are created together in CoordinatorsBuffer#find_or_create, and CoordinatorsBuffer#revoke only calls us once it has confirmed the coordinator exists - so the tracker is always present here.

Parameters:



41
42
43
 
# File 'lib/karafka/connection/pauses_manager.rb', line 41

def revoke(topic, partition)
  @pauses[topic][partition].reset
end