Add latest changes from gitlab-org/gitlab@13-8-stable-eev13.8.0-rc42

1 files changed, 13 insertions, 17 deletions

diff --git a/lib/bulk_imports/pipeline/runner.rb b/lib/bulk_imports/pipeline/runner.rb
index 88b96f0ab6e..11fb9722173 100644
--- a/lib/bulk_imports/pipeline/runner.rb
+++ b/lib/bulk_imports/pipeline/runner.rb
@@ -12,25 +12,15 @@ module BulkImports
 
         info(context, message: 'Pipeline started', pipeline_class: pipeline)
 
-        extractors.each do |extractor|
-          data = run_pipeline_step(:extractor, extractor.class.name, context) do
-            extractor.extract(context)
+        Array.wrap(extracted_data_from(context)).each do |entry|
+          transformers.each do |transformer|
+            entry = run_pipeline_step(:transformer, transformer.class.name, context) do
+              transformer.transform(context, entry)
+            end
           end
 
-          if data && data.respond_to?(:each)
-            data.each do |entry|
-              transformers.each do |transformer|
-                entry = run_pipeline_step(:transformer, transformer.class.name, context) do
-                  transformer.transform(context, entry)
-                end
-              end
-
-              loaders.each do |loader|
-                run_pipeline_step(:loader, loader.class.name, context) do
-                  loader.load(context, entry)
-                end
-              end
-            end
+          run_pipeline_step(:loader, loader.class.name, context) do
+            loader.load(context, entry)
           end
         end
 
@@ -55,6 +45,12 @@ module BulkImports
         mark_as_failed(context) if abort_on_failure?
       end
 
+      def extracted_data_from(context)
+        run_pipeline_step(:extractor, extractor.class.name, context) do
+          extractor.extract(context)
+        end
+      end
+
       def mark_as_failed(context)
         warn(context, message: 'Pipeline failed', pipeline_class: pipeline)