Class: Racecar::Runner

Inherits:
Object
  • Object
show all
Defined in:
lib/racecar/runner.rb

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(consumer_class, config:, logger:, instrumenter: NullInstrumenter) ⇒ Runner

Returns a new instance of Runner.



21
22
23
24
25
26
27
28
29
30
31
 
# File 'lib/racecar/runner.rb', line 21

def initialize(consumer_class, config:, logger:, instrumenter: NullInstrumenter)
  @consumer_class, @config, @logger = consumer_class, config, logger
  @instrumenter = instrumenter
  @stop_requested = false
  @partition_processors = Concurrent::Hash.new
  @consumer_class_instance = consumer_class.new
  if @consumer_class_instance.respond_to?(:statistics_callback) && Rdkafka::Config.statistics_callback.nil?
    Rdkafka::Config.statistics_callback = @consumer_class_instance.method(:statistics_callback).to_proc
  end
  Rdkafka::Config.logger = logger
end

Instance Attribute Details

#config ⇒ Object (readonly)

Returns the value of attribute config.



14
15
16
 
# File 'lib/racecar/runner.rb', line 14

def config
  @config
end

#consumer_class ⇒ Object (readonly)

Returns the value of attribute consumer_class.



14
15
16
 
# File 'lib/racecar/runner.rb', line 14

def consumer_class
  @consumer_class
end

#logger ⇒ Object (readonly)

Returns the value of attribute logger.



14
15
16
 
# File 'lib/racecar/runner.rb', line 14

def logger
  @logger
end

#partition_processors ⇒ Object (readonly)

Returns the value of attribute partition_processors.



14
15
16
 
# File 'lib/racecar/runner.rb', line 14

def partition_processors
  @partition_processors
end

Class Method Details

.topic_partition_key(topic, partition) ⇒ Object 



33
34
35
 
# File 'lib/racecar/runner.rb', line 33

def self.topic_partition_key(topic, partition)
  "#{topic}/#{partition}"
end

Instance Method Details

#consumer ⇒ Object 



99
100
101
102
103
 
# File 'lib/racecar/runner.rb', line 99

def consumer
  @consumer ||= begin
    ConsumerSet.new(config, logger, @partition_processors, @instrumenter)
  end
end

#processor ⇒ Object

Kept for backward compatibility — external code calls ‘processor`.



17
18
19
 
# File 'lib/racecar/runner.rb', line 17

def processor
  @consumer_class_instance
end

#run ⇒ Object 



37
38
39
40
41
42
43
44
45
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
83
84
85
86
87
88
89
90
91
92
93
 
# File 'lib/racecar/runner.rb', line 37

def run
  install_signal_handlers
  @stop_requested = false

  unless config.multithreaded_processing_enabled
    @consumer_class_instance.configure(
      producer:     consumer.producer,
      consumer:     consumer,
      instrumenter: @instrumenter,
      config:       config,
    )
  end

  loop_payload = {
    consumer_class: consumer_class.to_s,
    consumer_set: consumer
  }
  # Main loop
  begin
    loop do
      break if @stop_requested

      @instrumenter.instrument("start_main_loop", loop_payload)
      @instrumenter.instrument("main_loop", loop_payload) do
        resume_all_paused_partitions unless config.multithreaded_processing_enabled

        case process_method
        when :batch then
          msg_per_part = consumer.batch_poll(config.max_wait_time_ms).group_by(&:partition)
          msg_per_part.each_value do |messages_per_partition|
            processor = assign_and_get_processor(messages_per_partition)
            processor&.process_batch(messages_per_partition) unless processor&.rebalancing_or_shutting_down?
          end
        when :single then
          message = consumer.poll(config.max_wait_time_ms)
          if message
            processor = assign_and_get_processor(message)
            processor&.process(message) unless processor&.rebalancing_or_shutting_down?
          end
        end
      end
    end
  ensure
    logger.info "Gracefully shutting down"
    shutdown_processors_and_wait
    consumer.commit
  end
ensure
  begin
    @instrumenter.instrument('leave_group') do
      consumer.close
    end
  ensure
    Racecar::Datadog.close if config.datadog_enabled
    @instrumenter.instrument("shut_down", loop_payload || {})
  end
end

#stop ⇒ Object 



95
96
97
 
# File 'lib/racecar/runner.rb', line 95

def stop
  @stop_requested = true
end