You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: executor.rb
    path of file: ./repos/karafka/lib/karafka/processing
    
    the code of the file until where you have to start completion: # frozen_string_literal: true

module Karafka
  # Namespace that encapsulates all the logic related to processing data.
  module Processing
    # Executors:
    # - run consumers code (for `#call`) or run given preparation / teardown operations when needed
    #   from separate threads.
    # - they re-create consumer instances in case of partitions that were revoked and assigned
    #   back.
    #
    # @note Executors are not removed after partition is revoked. They are not that big and will
    #   be re-used in case of a re-claim
    #
    # @note Since given consumer can run various operations, executor manages that and its
    #   lifecycle. There are following types of operations with appropriate before/after, etc:
    #
    #   - consume - primary operation related to running user consumption code
    #   - idle - cleanup job that runs on idle runs where no messages would be passed to the end
    #     user. This is used fo
    
      def shutdown
        # There is a case, where the consumer no longer exists because it was revoked, in case like
        # that we do not build a new instance and shutdown should not be triggered.
        consumer.on_shutdown if @consumer
      end

      private

      # @return [Object] cached consumer instance
      def consumer
        @consumer ||= begin
          topic = @coordinator.topic

          strategy = strategy_selector.find(topic)
          expansions = expansions_selector.find(topic)

          consumer = topic.consumer_class.new
          # We use singleton class as the same consumer class may be used to process different
          # topics with different settings
          consumer.singleton_class.include(strategy)

          # Specific features may expand consumer API beyond the injected strategy. The difference
          # here is that strategy impacts the flow of states while extra APIs just provide some
          # extra methods with informations, etc but do no deviate the flow behavior
          expansions.each { |expansion| consumer.singleton_class.include(expansion) }

          consumer.client = @client
          consumer.producer = ::Karafka::App.producer
          consumer.coordinator = @coordinator
          # Since we have some message-less flows (idle, etc), we initialize consumer with empty
          # messages set. In production we have persistent consumers, so this is not a performance
          # overhead as this will happen only once per consumer lifetime
          consumer.messages = empty_messages

          consumer
        end
      end

      # Initializes the messages set in case given operation would happen before any processing
      # This prevents us from having no messages object at all as the messages object and
      # its metadata may be used for statistics
      def empty_messages
        Messages::Builders::Messages.call(
          [],
          topic,
          partition,
          Time.now
        )
      end
    end