Contents

# frozen_string_literal: true

module Karafka
  module Web
    module Processing
      module Consumers
        # Fetches the current consumer processes aggregated state
        class State
          class << self
            # Fetch the current consumers state that is expected to exist
            #
            # @return [Hash] last (current) aggregated processes state
            def current!
              state_message = ::Karafka::Admin.read_topic(
                Karafka::Web.config.topics.consumers.states,
                0,
                1
              ).last

              return state_message.payload if state_message

              raise(::Karafka::Web::Errors::Processing::MissingConsumersStateError)
            rescue Rdkafka::RdkafkaError => e
              raise(e) unless e.code == :unknown_partition

              raise(::Karafka::Web::Errors::Processing::MissingConsumersStatesTopicError)
            end
          end
        end
      end
    end
  end
end

Version data entries

3 entries across 3 versions & 1 rubygems

Version	Path
karafka-web-0.7.7	lib/karafka/web/processing/consumers/state.rb
karafka-web-0.7.6	lib/karafka/web/processing/consumers/state.rb
karafka-web-0.7.5	lib/karafka/web/processing/consumers/state.rb