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

#initialize ⇒ Object

This method calls the parent class initializer and then sets up the extra instance variable to nil. The explicit initialization to nil is included as an optimization for Ruby’s object shapes system, which improves memory layout and access performance.



34
35
36
37
 
# File 'lib/karafka/pro/routing/features/pausing/topic.rb', line 34

def initialize(...)
  super
  @pausing = nil
end

#pause(timeout: nil, max_timeout: nil, with_exponential_backoff: nil) ⇒ Config

Allows for per-topic pausing strategy setting

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:

  • (Config) —

    pausing config object



46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/karafka/pro/routing/features/pausing/topic.rb', line 46

def pause(timeout: nil, max_timeout: nil, with_exponential_backoff: nil)
  # If no arguments provided, just return or initialize the config
  return pausing if timeout.nil? && max_timeout.nil? && with_exponential_backoff.nil?

  # Update instance variables for backwards compatibility
  # This ensures code reading @pause_timeout directly or via the inherited getter
  # will get the correct values
  @pause_timeout = timeout if timeout
  @pause_max_timeout = max_timeout if max_timeout

  unless with_exponential_backoff.nil?
    @pause_with_exponential_backoff = with_exponential_backoff
  end

  # Create or update the config
  @pausing ||= Config.new(
    active: false,
    timeout: @pause_timeout || Karafka::App.config.pause.timeout,
    max_timeout: @pause_max_timeout || Karafka::App.config.pause.max_timeout,
    with_exponential_backoff: if @pause_with_exponential_backoff.nil?
                                Karafka::App.config.pause.with_exponential_backoff
                              else
                                @pause_with_exponential_backoff
                              end
  )

  @pausing.timeout = timeout if timeout
  @pausing.max_timeout = max_timeout if max_timeout

  unless with_exponential_backoff.nil?
    @pausing.with_exponential_backoff = with_exponential_backoff
  end

  @pausing.active = true

  @pausing
end

#pausing ⇒ Config

Returns pausing configuration object.

Returns:

  • (Config) —

    pausing configuration object



85
86
87
88
89
90
91
92
93
94
95
96
 
# File 'lib/karafka/pro/routing/features/pausing/topic.rb', line 85

def pausing
  @pausing ||= Config.new(
    active: false,
    timeout: @pause_timeout || Karafka::App.config.pause.timeout,
    max_timeout: @pause_max_timeout || Karafka::App.config.pause.max_timeout,
    with_exponential_backoff: if @pause_with_exponential_backoff.nil?
                                Karafka::App.config.pause.with_exponential_backoff
                              else
                                @pause_with_exponential_backoff
                              end
  )
end

#pausing? ⇒ Boolean

Returns is pausing explicitly configured.

Returns:

  • (Boolean) —

    is pausing explicitly configured



99
100
101
 
# File 'lib/karafka/pro/routing/features/pausing/topic.rb', line 99

def pausing?
  pausing.active?
end

#to_h ⇒ Hash

Returns topic with all its native configuration options plus pausing settings.

Returns:

  • (Hash) —

    topic with all its native configuration options plus pausing settings



104
105
106
107
108
 
# File 'lib/karafka/pro/routing/features/pausing/topic.rb', line 104

def to_h
  super.merge(
    pausing: pausing.to_h
  ).freeze
end