Class: Karafka::Pro::Processing::ConsumerGroups::Coordinators::ErrorsTracker

Inherits:
Object
  • Object
show all
Includes:
Enumerable
Defined in:
lib/karafka/pro/processing/consumer_groups/coordinators/errors_tracker.rb

Overview

Object used to track errors in between executions to be able to build error-type based recovery flows.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(topic, partition, limit: STORAGE_LIMIT) ⇒ ErrorsTracker

Note:

limit does not apply to the counts. They will work beyond the number of errors occurring

Returns a new instance of ErrorsTracker.

Parameters:

  • topic (Karafka::Routing::Topic)
  • partition (Integer)
  • limit (Integer) (defaults to: STORAGE_LIMIT) —

    max number of errors we want to keep for reference when implementing custom error handling.



67
68
69
70
71
72
73
74
 
# File 'lib/karafka/pro/processing/consumer_groups/coordinators/errors_tracker.rb', line 67

def initialize(topic, partition, limit: STORAGE_LIMIT)
  @errors = []
  @counts = Hash.new { |hash, key| hash[key] = 0 }
  @topic = topic
  @partition = partition
  @limit = limit
  @trace_id = SecureRandom.uuid
end

Instance Attribute Details

#counts ⇒ Hash (readonly)

Returns:

  • (Hash) 


49
50
51
 
# File 'lib/karafka/pro/processing/consumer_groups/coordinators/errors_tracker.rb', line 49

def counts
  @counts
end

#partition ⇒ Integer (readonly)

Returns partition of this error tracker.

Returns:

  • (Integer) —

    partition of this error tracker



46
47
48
 
# File 'lib/karafka/pro/processing/consumer_groups/coordinators/errors_tracker.rb', line 46

def partition
  @partition
end

#topic ⇒ Karafka::Routing::Topic (readonly)

Returns topic of this error tracker.

Returns:



43
44
45
 
# File 'lib/karafka/pro/processing/consumer_groups/coordinators/errors_tracker.rb', line 43

def topic
  @topic
end

#trace_id ⇒ String (readonly)

Returns:

  • (String) 


52
53
54
 
# File 'lib/karafka/pro/processing/consumer_groups/coordinators/errors_tracker.rb', line 52

def trace_id
  @trace_id
end

Instance Method Details

#<<(error) ⇒ Object

Parameters:

  • error (StandardError) —

    adds the error to the tracker



83
84
85
86
87
88
 
# File 'lib/karafka/pro/processing/consumer_groups/coordinators/errors_tracker.rb', line 83

def <<(error)
  @errors.shift if @errors.size >= @limit
  @errors << error
  @counts[error.class] += 1
  @trace_id = SecureRandom.uuid
end

#all ⇒ Array<StandardError>

Returns array with all the errors that occurred.

Returns:

  • (Array<StandardError>) —

    array with all the errors that occurred



113
114
115
 
# File 'lib/karafka/pro/processing/consumer_groups/coordinators/errors_tracker.rb', line 113

def all
  @errors
end

#clear ⇒ Object

Clears all the errors



77
78
79
80
 
# File 'lib/karafka/pro/processing/consumer_groups/coordinators/errors_tracker.rb', line 77

def clear
  @errors.clear
  @counts.clear
end

#each ⇒ Object

Iterates over errors



108
109
110
 
# File 'lib/karafka/pro/processing/consumer_groups/coordinators/errors_tracker.rb', line 108

def each(&)
  @errors.each(&)
end

#empty? ⇒ Boolean

Returns is the error tracker empty.

Returns:

  • (Boolean) —

    is the error tracker empty



91
92
93
 
# File 'lib/karafka/pro/processing/consumer_groups/coordinators/errors_tracker.rb', line 91

def empty?
  @errors.empty?
end

#last ⇒ StandardError?

Returns last error that occurred or nil if no errors.

Returns:

  • (StandardError, nil) —

    last error that occurred or nil if no errors



103
104
105
 
# File 'lib/karafka/pro/processing/consumer_groups/coordinators/errors_tracker.rb', line 103

def last
  @errors.last
end

#size ⇒ Integer

Returns number of elements.

Returns:

  • (Integer) —

    number of elements



96
97
98
99
100
 
# File 'lib/karafka/pro/processing/consumer_groups/coordinators/errors_tracker.rb', line 96

def size
  # We use counts reference of all errors and not the `@errors` array because it allows
  # us to go beyond the whole errors storage limit
  @counts.values.sum
end