Module: Karafka::Pro::Routing::Features::Pausing::Topic

Defined in:
lib/karafka/pro/routing/features/pausing/topic.rb

Overview

Expansion allowing for a per topic pause strategy definitions

Instance Method Summary collapse

Instance Method Details

#pause(timeout: nil, max_timeout: nil, with_exponential_backoff: nil) ⇒ Karafka::Routing::Features::Pausing::Config

Allows for per-topic pausing strategy setting.

Overrides the OSS #pause reader (this module is prepended onto Routing::Topic). With no arguments it returns the current configuration, defaulting to the global config.pause.* settings via super. With arguments it overrides the pausing strategy for this topic and marks it as active.

Parameters:

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

    how long should we wait upon processing error (milliseconds)

  • max_timeout (Integer) (defaults to: nil) —

    what is the max timeout in case of an exponential backoff (milliseconds)

  • with_exponential_backoff (Boolean) (defaults to: nil) —

    should we use exponential backoff

Returns:



50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/karafka/pro/routing/features/pausing/topic.rb', line 50

def pause(timeout: nil, max_timeout: nil, with_exponential_backoff: nil)
  config = super()

  # If no arguments provided, just return the current (default or overridden) config
  return config if timeout.nil? && max_timeout.nil? && with_exponential_backoff.nil?

  config.timeout = timeout if timeout
  config.max_timeout = max_timeout if max_timeout

  unless with_exponential_backoff.nil?
    config.with_exponential_backoff = with_exponential_backoff
  end

  config.active = true

  config
end

#pause? ⇒ Boolean

Returns is pausing explicitly configured on a per-topic basis.

Returns:

  • (Boolean) —

    is pausing explicitly configured on a per-topic basis



69
70
71
 
# File 'lib/karafka/pro/routing/features/pausing/topic.rb', line 69

def pause?
  pause.active?
end