FIX: missed exception in plugin error (#4945)

* FIX: missed exception in plugin error building

Signed-off-by: George Chen <[email protected]>

data-prepper-core/src/main/java/org/opensearch/dataprepper/core/parser/PipelineTransformer.java

@@ -157,6 +157,7 @@ private void buildPipelineFromConfiguration(
                         .componentType(PipelineModel.BUFFER_PLUGIN_TYPE)
                         .pipelineName(pipelineName)
                         .pluginName(bufferPluginSetting.getName())
+                        .exception(e)
                         .build();
                 pluginErrorCollector.collectPluginError(pluginError);
             }

data-prepper-core/src/test/java/org/opensearch/dataprepper/TestDataProvider.java

@@ -16,6 +16,9 @@ public class TestDataProvider {
     public static final String VALID_OFF_HEAP_FILE_WITH_ACKS = "src/test/resources/multiple_pipeline_valid_off_heap_buffer_with_acks.yml";
     public static final String DISCONNECTED_VALID_OFF_HEAP_FILE_WITH_ACKS = "src/test/resources/multiple_disconnected_pipeline_valid_off_heap_buffer_with_acks.yml";
     public static final String CONNECTED_PIPELINE_ROOT_SOURCE_INCORRECT = "src/test/resources/connected_pipeline_incorrect_root_source.yml";
+    public static final String CONNECTED_PIPELINE_BUFFER_INCORRECT = "src/test/resources/connected_pipeline_incorrect_buffer.yml";
+    public static final String CONNECTED_PIPELINE_PROCESSOR_INCORRECT = "src/test/resources/connected_pipeline_incorrect_processor.yml";
+    public static final String CONNECTED_PIPELINE_SINK_INCORRECT = "src/test/resources/connected_pipeline_incorrect_sink.yml";
     public static final String CONNECTED_PIPELINE_CHILD_PIPELINE_INCORRECT_DUE_TO_SINK = "src/test/resources/connected_pipeline_incorrect_child_pipeline_due_to_invalid_sink.yml";
     public static final String CONNECTED_PIPELINE_CHILD_PIPELINE_INCORRECT_DUE_TO_PROCESSOR = "src/test/resources/connected_pipeline_incorrect_child_pipeline_due_to_invalid_processor.yml";
     public static final String CYCLE_MULTIPLE_PIPELINE_CONFIG_FILE = "src/test/resources/cyclic_multiple_pipeline_configuration.yml";

data-prepper-core/src/test/java/org/opensearch/dataprepper/core/parser/PipelineTransformerTests.java

@@ -215,18 +215,37 @@ void parseConfiguration_with_multiple_disconnected_valid_pipelines_creates_the_c
         assertThat(pipeline.getSource().areAcknowledgementsEnabled(),equalTo(false));
     }
 
-
     @Test
-    void parseConfiguration_with_invalid_root_pipeline_creates_empty_pipelinesMap() {
+    void parseConfiguration_with_invalid_root_source_pipeline_creates_empty_pipelinesMap() {
         final PipelineTransformer pipelineTransformer =
                 createObjectUnderTest(TestDataProvider.CONNECTED_PIPELINE_ROOT_SOURCE_INCORRECT);
         final Map<String, Pipeline> connectedPipelines = pipelineTransformer.transformConfiguration();
         assertThat(connectedPipelines.size(), equalTo(0));
         verify(dataPrepperConfiguration).getPipelineExtensions();
+        assertThat(pluginErrorCollector.getPluginErrors().size(), equalTo(2));
+        final PluginError sourcePluginError = pluginErrorCollector.getPluginErrors().get(0);
+        assertThat(sourcePluginError.getPipelineName(), equalTo("test-pipeline-1"));
+        assertThat(sourcePluginError.getPluginName(), equalTo("file"));
+        assertThat(sourcePluginError.getException(), notNullValue());
+        // Buffer plugin gets error due to instantiated source is null
+        final PluginError bufferPluginError = pluginErrorCollector.getPluginErrors().get(1);
+        assertThat(bufferPluginError.getPipelineName(), equalTo("test-pipeline-1"));
+        assertThat(bufferPluginError.getPluginName(), equalTo("bounded_blocking"));
+        assertThat(bufferPluginError.getException(), notNullValue());
+    }
+
+    @ParameterizedTest
+    @MethodSource("provideInvalidPipelineResourceAndFailedPluginNameArgs")
+    void parseConfiguration_with_invalid_root_pipeline_creates_empty_pipelinesMap(
+            final String pipelineResourcePath, final String failedPluginName) {
+        final PipelineTransformer pipelineTransformer = createObjectUnderTest(pipelineResourcePath);
+        final Map<String, Pipeline> connectedPipelines = pipelineTransformer.transformConfiguration();
+        assertThat(connectedPipelines.size(), equalTo(0));
+        verify(dataPrepperConfiguration).getPipelineExtensions();
         assertThat(pluginErrorCollector.getPluginErrors().size(), equalTo(1));
         final PluginError pluginError = pluginErrorCollector.getPluginErrors().get(0);
         assertThat(pluginError.getPipelineName(), equalTo("test-pipeline-1"));
-        assertThat(pluginError.getPluginName(), equalTo("file"));
+        assertThat(pluginError.getPluginName(), equalTo(failedPluginName));
         assertThat(pluginError.getException(), notNullValue());
     }
 
@@ -562,6 +581,14 @@ private static Stream<Arguments> provideGetSecondaryBufferArgs() {
         );
     }
 
+    private static Stream<Arguments> provideInvalidPipelineResourceAndFailedPluginNameArgs() {
+        return Stream.of(
+                Arguments.of(TestDataProvider.CONNECTED_PIPELINE_BUFFER_INCORRECT, "invalid_buffer"),
+                Arguments.of(TestDataProvider.CONNECTED_PIPELINE_PROCESSOR_INCORRECT, "invalid_processor"),
+                Arguments.of(TestDataProvider.CONNECTED_PIPELINE_SINK_INCORRECT, "invalid_sink")
+        );
+    }
+
     private Map<String, Map<String, PeerForwarderReceiveBuffer<Record<Event>>>> generateBufferMap(final int outerMapEntryCount, final int innerMapEntryCount) {
         final Map<String, Map<String, PeerForwarderReceiveBuffer<Record<Event>>>> bufferMap = new HashMap<>();
         for (int i = 0; i < outerMapEntryCount; i++) {

data-prepper-core/src/test/resources/connected_pipeline_incorrect_buffer.yml

@@ -0,0 +1,23 @@
+# this configuration file is solely for testing formatting
+test-pipeline-1:
+  source:
+    stdin:
+  buffer:
+    invalid_buffer: #this plugin fails
+  sink:
+    - pipeline:
+        name: "test-pipeline-2"
+test-pipeline-2:
+  source:
+    pipeline:
+      name: "test-pipeline-1"
+  sink:
+    - pipeline:
+        name: "test-pipeline-3"
+test-pipeline-3:
+  source:
+    pipeline:
+      name: "test-pipeline-2"
+  sink:
+    - file:
+        path: "/tmp/todelete.txt"

data-prepper-core/src/test/resources/connected_pipeline_incorrect_processor.yml

@@ -0,0 +1,23 @@
+# this configuration file is solely for testing formatting
+test-pipeline-1:
+  source:
+    stdin:
+  processor:
+    - invalid_processor: # this plugin should fail
+  sink:
+    - pipeline:
+        name: "test-pipeline-2"
+test-pipeline-2:
+  source:
+    pipeline:
+      name: "test-pipeline-1"
+  sink:
+    - pipeline:
+        name: "test-pipeline-3"
+test-pipeline-3:
+  source:
+    pipeline:
+      name: "test-pipeline-2"
+  sink:
+    - file:
+        path: "/tmp/todelete.txt"

data-prepper-core/src/test/resources/connected_pipeline_incorrect_sink.yml

@@ -0,0 +1,22 @@
+# this configuration file is solely for testing formatting
+test-pipeline-1:
+  source:
+    stdin:
+  sink:
+    - pipeline:
+        name: "test-pipeline-2"
+    - invalid_sink: # this plugin should fail
+test-pipeline-2:
+  source:
+    pipeline:
+      name: "test-pipeline-1"
+  sink:
+    - pipeline:
+        name: "test-pipeline-3"
+test-pipeline-3:
+  source:
+    pipeline:
+      name: "test-pipeline-2"
+  sink:
+    - file:
+        path: "/tmp/todelete.txt"