Added Null validation and Custom Exception along with few more Test cases

Signed-off-by: Pallempati Saketh <pallempati.saketh@fmr.com>

Author: saketh-pallempati

data-prepper-pipeline-parser/src/main/java/org/opensearch/dataprepper/pipeline/parser/InvalidPipelineConfigurationException.java

@@ -0,0 +1,14 @@
+package org.opensearch.dataprepper.pipeline.parser;
+
+/**
+ * Exception thrown when pipeline configuration validation fails.
+ */
+public class InvalidPipelineConfigurationException extends RuntimeException {
+    public InvalidPipelineConfigurationException(String message) {
+        super(message);
+    }
+
+    public InvalidPipelineConfigurationException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

data-prepper-pipeline-parser/src/main/java/org/opensearch/dataprepper/pipeline/parser/PipelineConfigurationValidator.java

@@ -9,6 +9,7 @@
 import org.opensearch.dataprepper.model.configuration.PluginSetting;
 import org.opensearch.dataprepper.pipeline.parser.model.PipelineConfiguration;
 import org.opensearch.dataprepper.pipeline.parser.model.SinkContextPluginSetting;
+import org.opensearch.dataprepper.pipeline.parser.InvalidPipelineConfigurationException;
 import org.opensearch.dataprepper.model.configuration.ConditionalRoute;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -194,12 +195,15 @@ private static void validateSinkRoutes(final Map<String, PipelineConfiguration>
             final List<SinkContextPluginSetting> sinkSettings = pipelineConfiguration.getSinkPluginSettings();
             for (SinkContextPluginSetting sinkPlugin : sinkSettings) {
                 Collection<String> sinkRoutes = sinkPlugin.getSinkContext().getRoutes();
+                if (sinkRoutes == null) {
+                    sinkRoutes = Collections.emptyList();
+                }
                 List<String> invalidRoutes = sinkRoutes.stream()
                         .filter(route -> !validRoutes.contains(route))
                         .collect(Collectors.toList());
 
                 if (!invalidRoutes.isEmpty()) {
-                    throw new RuntimeException(String.format(
+                    throw new InvalidPipelineConfigurationException(String.format(
                             "The following routes do not exist in pipeline \"%s\": %s. Configured routes include %s",
                             pipelineName, invalidRoutes, validRoutes));
                 }

data-prepper-pipeline-parser/src/test/java/org/opensearch/dataprepper/pipeline/parser/PipelineConfigurationValidatorIT.java

@@ -16,6 +16,10 @@
 import java.util.Map;
 import java.util.stream.Collectors;
 
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.hasSize;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -39,23 +43,74 @@ public class PipelineConfigurationValidatorIT {
                     "        routes:\n" +
                     "          - SECOND_ROUTE\n";
 
+    private static final String SIMPLE_PIPELINE_NO_ROUTES_CONFIG =
+            "entry-pipeline:\n" +
+                    "  source:\n" +
+                    "    random:\n" +
+                    "  sink:\n" +
+                    "    - stdout:\n";
+
+    private static final String ROUTES_DEFINED_NO_SINK_ROUTES_CONFIG =
+            "entry-pipeline:\n" +
+                    "  source:\n" +
+                    "    random:\n" +
+                    "  routes:\n" +
+                    "    - route_one: \"/my_route == 10\"\n" +
+                    "    - route_two: \"/my_route == 11\"\n" +
+                    "  sink:\n" +
+                    "    - stdout:\n";
+
     @Test
     public void test_with_invalid_sink_routes_in_yaml_should_throw() {
+        Map<String, PipelineConfiguration> pipelineConfigMap = getPipelineConfigMap(TEST_YAML_CONFIG);
+
+        InvalidPipelineConfigurationException exception = assertThrows(InvalidPipelineConfigurationException.class, () ->
+                PipelineConfigurationValidator.validateAndGetPipelineNames(pipelineConfigMap));
+
+        String expectedErrorMessage = "The following routes do not exist in pipeline \"entry-pipeline\": " +
+                "[FIRST_ROUTE]. Configured routes include [route_one, route_two]";
+
+        assertThat(exception.getMessage(), equalTo(expectedErrorMessage));
+    }
+
+    @Test
+    public void test_simple_pipeline_with_no_routes_succeeds() {
+        Map<String, PipelineConfiguration> pipelineConfigMap = getPipelineConfigMap(SIMPLE_PIPELINE_NO_ROUTES_CONFIG);
+
+        List<String> pipelineNames = assertDoesNotThrow(() ->
+                PipelineConfigurationValidator.validateAndGetPipelineNames(pipelineConfigMap));
+
+        assertThat(pipelineNames, hasSize(1));
+        assertThat(pipelineNames.get(0), equalTo("entry-pipeline"));
+    }
+
+    @Test
+    public void test_routes_defined_but_no_routes_on_sinks_succeeds() {
+        Map<String, PipelineConfiguration> pipelineConfigMap = getPipelineConfigMap(ROUTES_DEFINED_NO_SINK_ROUTES_CONFIG);
+
+        List<String> pipelineNames = assertDoesNotThrow(() ->
+                PipelineConfigurationValidator.validateAndGetPipelineNames(pipelineConfigMap));
+
+        assertThat(pipelineNames, hasSize(1));
+        assertThat(pipelineNames.get(0), equalTo("entry-pipeline"));
+    }
+
+    /**
+     * Helper method to create a pipeline configuration map from a YAML string
+     */
+    private Map<String, PipelineConfiguration> getPipelineConfigMap(final String yamlConfig) {
         PipelineConfigurationReader configReader = mock(PipelineConfigurationReader.class);
         when(configReader.getPipelineConfigurationInputStreams())
-                .thenReturn(List.of(new ByteArrayInputStream(TEST_YAML_CONFIG.getBytes())));
+                .thenReturn(List.of(new ByteArrayInputStream(yamlConfig.getBytes())));
 
         PipelinesDataflowModelParser modelParser = new PipelinesDataflowModelParser(configReader);
         PipelinesDataFlowModel dataFlowModel = modelParser.parseConfiguration();
 
-        Map<String, PipelineConfiguration> pipelineConfigMap = dataFlowModel.getPipelines().entrySet()
+        return dataFlowModel.getPipelines().entrySet()
                 .stream()
                 .collect(Collectors.toMap(
                         Map.Entry::getKey,
                         entry -> new PipelineConfiguration(entry.getValue())
                 ));
-
-        assertThrows(RuntimeException.class, () ->
-                PipelineConfigurationValidator.validateAndGetPipelineNames(pipelineConfigMap));
     }
 }