Class: Kafka::Pause

Inherits:
Object
  • Object
show all
Defined in:
lib/kafka/pause.rb

Overview

Manages the pause state of a partition.

The processing of messages in a partition can be paused, e.g. if there was an exception during processing. This could be caused by a downstream service not being available. A typical way of solving such an issue is to back off for a little while and then try again. In order to do that, pause the partition.

Instance Method Summary collapse

Constructor Details

#initialize(clock: Time) ⇒ Pause

Returns a new instance of Pause.

 12 13 14 15 16 17 18 19
 
# File 'lib/kafka/pause.rb', line 12 def initialize(clock: Time) @clock = clock @started_at = nil @pauses = 0 @timeout = nil @max_timeout = nil @exponential_backoff = false end

Instance Method Details

#expired? ⇒ Boolean

Whether the pause has expired.

Returns:

  • (Boolean) 
 66 67 68 69 70 71 72
 
# File 'lib/kafka/pause.rb', line 66 def expired? # We never expire the pause if timeout is nil. return false if @timeout.nil? # Have we passed the end of the pause duration? @clock.now >= ends_at end

#pause!(timeout: nil, max_timeout: nil, exponential_backoff: false) ⇒ Object

Mark the partition as paused.

If exponential backoff is enabled, each subsequent pause of a partition will cause a doubling of the actual timeout, i.e. for pause number n, the actual timeout will be 2^n * timeout.

Only when #reset! is called is this state cleared.

Parameters:

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

    if specified, the partition will automatically resume after this many seconds.

  • exponential_backoff (Boolean) (defaults to: false) —

    whether to enable exponential timeouts.

 32 33 34 35 36 37 38
 
# File 'lib/kafka/pause.rb', line 32 def pause!(timeout: nil, max_timeout: nil, exponential_backoff: false) @started_at = @clock.now @timeout = timeout @max_timeout = max_timeout @exponential_backoff = exponential_backoff @pauses += 1 end

#pause_duration ⇒ Object 

 57 58 59 60 61 62 63
 
# File 'lib/kafka/pause.rb', line 57 def pause_duration if paused? Time.now - @started_at else 0 end end

#paused? ⇒ Boolean

Whether the partition is currently paused. The pause may have expired, in which case #expired? should be checked as well.

Returns:

  • (Boolean) 
 52 53 54 55
 
# File 'lib/kafka/pause.rb', line 52 def paused? # This is nil if we're not currently paused. !@started_at.nil? end

#reset! ⇒ Object

Resets the pause state, ensuring that the next pause is not exponential.

 75 76 77
 
# File 'lib/kafka/pause.rb', line 75 def reset! @pauses = 0 end

#resume! ⇒ Object

Resumes the partition.

The number of pauses is still retained, and if the partition is paused again it may be with an exponential backoff.

 44 45 46 47 48
 
# File 'lib/kafka/pause.rb', line 44 def resume! @started_at = nil @timeout = nil @max_timeout = nil end