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: runner.rb
    path of file: ./repos/gitlabhq/lib/bulk_imports/pipeline
    
    the code of the file until where you have to start completion: # frozen_string_literal: true

module BulkImports
  module Pipeline
    module Runner
      extend ActiveSupport::Concern

      MarkedAsFailedError = Class.new(StandardError)

      def run
        raise MarkedAsFailedError if context.entity.failed?

        info(message: 'Pipeline started')

        set_source_objects_counter
        extracted_data = extracted_data_from

        if extracted_data
          extracted_data.each_with_index do |entry, index|
            refresh_entity_and_import if index % 1000 == 0

            raw_entry = entry.dup
            next if already_processed?(raw_entry, index)

            increment_fetched_objects_counter

            transformers.each do |transformer|
              entry = run_pipeline_step(:transformer, transformer.class.name) do
                transformer.transform(context, entry)
              end
            end

            run_pipeline_step(:loader, loader.class.name, entry) do
              loader.load(context, entry)

              increment_imported_objects_counter
            end

            save_processed_entry(raw_entry, index)
          end

          tracker.update!(
            has_next_page: extracted_data.has_next_page?,
            next_page: extracted_data.next_page
          )

          run_pipeline_step(:after_run) do
            after_run(extracted_data)
          end

          # For batches, `#on_finish` is called once within `FinishBatchedPipelineWorker`
          # after all batches have completed.
          unless tracker.batched?
            run_pipeline_step(:on_finish) do
              on_finish
            end
          end
        end

        info(message: 'Pipeline finished')
      rescue MarkedAsFailedError
        skip!('Skipping pipeline due to failed entity')
      end

      def on_finish; end

      private # rubocop:disable Lint/UselessAccessModifier

      def run_pipeline_step(step, class_name = nil, entry = nil)
        raise MarkedAsFailedError if context.entity.failed?

        info(pipeline_step: step, step_class: class_name)

        yield
      rescue MarkedAsFailedError
        skip!(
          'Skipping pipeline due to failed entity',
          pipeline_step: step,
          step_class: class_name,
          importer: 'gitlab_migration'
        )
      rescue BulkImports::NetworkError => e
        raise BulkImports::RetryPipelineError.new(e.message, e.retry_delay) if e.retriable?(context.tracker)

        log_and_fail(e, step, entry)
      rescue BulkImports::RetryPipelineError
        raise
      rescue StandardError => e
        log_and_fail(e, step, entry)
      end

      def extracted_data_from
        run_pipeline_step(:extractor, extractor.class.name) do
         
    
      def on_finish; end

      private # rubocop:disable Lint/UselessAccessModifier

      def run_pipeline_step(step, class_name = nil, entry = nil)
        raise MarkedAsFailedError if context.entity.failed?

        info(pipeline_step: step, step_class: class_name)

        yield
      rescue MarkedAsFailedError
        skip!(
          'Skipping pipeline due to failed entity',
          pipeline_step: step,
          step_class: class_name,
          importer: 'gitlab_migration'
        )
      rescue BulkImports::NetworkError => e
        raise BulkImports::RetryPipelineError.new(e.message, e.retry_delay) if e.retriable?(context.tracker)

        log_and_fail(e, step, entry)
      rescue BulkImports::RetryPipelineError
        raise
      rescue StandardError => e
        log_and_fail(e, step, entry)
      end

      def extracted_data_from
        run_pipeline_step(:extractor, extractor.class.name) do
          extractor.extract(context)
        end
      end

      def cache_key
        batch_number = context.extra[:batch_number] || 0

        "#{self.class.name.underscore}/#{tracker.bulk_import_entity_id}/#{batch_number}"
      end

      # Overridden by child pipelines with different caching strategies
      def already_processed?(*)
        false
      end

      def save_processed_entry(*); end

      def after_run(extracted_data)
        run if extracted_data.has_next_page?
      end

      def log_and_fail(exception, step, entry = nil)
        log_import_failure(exception, step, entry)

        if abort_on_failure?
          tracker.fail_op!

          warn(message: 'Aborting entity migration due to pipeline failure')
          context.entity.fail_op!
        end

        nil
      end

      def skip!(message, extra = {})
        warn({ message: message }.merge(extra))

        tracker.skip!
      end

      def log_import_failure(exception, step, entry)
        failure_attributes = {
          bulk_import_entity_id: context.entity.id,
          pipeline_class: pipeline,
          pipeline_step: step,
          exception_class: exception.class.to_s,
          exception_message: exception.message,
          correlation_id_value: Labkit::Correlation::CorrelationId.current_or_new_id
        }

        if entry
          failure_attributes[:source_url] = BulkImports::SourceUrlBuilder.new(context, entry).url
          failure_attributes[:source_title] = entry.try(:title) || entry.try(:name)
        end

        log_exception(
          exception,
          log_params(
            {
              bulk_import_id: context.bulk_import_id,
              pipeline_step: step,
              message: 'An object of a pipeline failed to import'
            }
          )
        )

        BulkImports::Failure.create(failure_attributes)
      end