Validate plugin configuration objects using JSR-303 validation

data-prepper-api/src/main/java/com/amazon/dataprepper/model/plugin/InvalidPluginConfigurationException.java

@@ -0,0 +1,18 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package com.amazon.dataprepper.model.plugin;
+
+/**
+ * This exception is thrown when a plugin in configured with an invalid
+ * configuration.
+ *
+ * @since 1.3
+ */
+public class InvalidPluginConfigurationException extends RuntimeException {
+    public InvalidPluginConfigurationException(final String message) {
+        super(message);
+    }
+}

data-prepper-api/src/test/java/com/amazon/dataprepper/model/plugin/InvalidPluginConfigurationExceptionTest.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package com.amazon.dataprepper.model.plugin;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.UUID;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+class InvalidPluginConfigurationExceptionTest {
+    private String message;
+
+    @BeforeEach
+    void setUp() {
+        message = UUID.randomUUID().toString();
+    }
+
+    private InvalidPluginConfigurationException createObjectUnderTest() {
+        return new InvalidPluginConfigurationException(message);
+    }
+
+    @Test
+    void getMessage_returns_message() {
+        assertThat(createObjectUnderTest().getMessage(),
+                equalTo(message));
+    }
+}

data-prepper-core/build.gradle

@@ -20,13 +20,13 @@ dependencies {
     implementation 'com.fasterxml.jackson.core:jackson-databind'
     implementation 'com.fasterxml.jackson.dataformat:jackson-dataformat-yaml'
     implementation "javax.validation:validation-api:2.0.1.Final"
-    implementation "org.apache.bval:bval-extras:2.0.5"
-    implementation "org.apache.bval:bval-jsr:2.0.5"
     implementation "org.reflections:reflections:0.10.2"
     implementation 'io.micrometer:micrometer-core'
     implementation 'io.micrometer:micrometer-registry-prometheus'
     implementation 'io.micrometer:micrometer-registry-cloudwatch2'
     implementation 'software.amazon.awssdk:cloudwatch'
+    implementation 'org.hibernate.validator:hibernate-validator:7.0.2.Final'
+    implementation 'org.glassfish:jakarta.el:4.0.1'
     implementation platform('org.apache.logging.log4j:log4j-bom:2.17.1')
     implementation 'org.apache.logging.log4j:log4j-core'
     implementation 'org.apache.logging.log4j:log4j-slf4j-impl'

data-prepper-core/src/main/java/com/amazon/dataprepper/plugin/PluginConfigurationConverter.java

@@ -5,26 +5,65 @@
 
 package com.amazon.dataprepper.plugin;
 
+import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.configuration.PluginSetting;
+import com.amazon.dataprepper.model.plugin.InvalidPluginConfigurationException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.PropertyNamingStrategies;
+import jakarta.validation.ConstraintViolation;
+import jakarta.validation.Validation;
+import jakarta.validation.Validator;
+import jakarta.validation.ValidatorFactory;
 
 import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
 
+/**
+ * Converts and validates a plugin configuration. This class is responsible for taking a {@link PluginSetting}
+ * and converting it to the plugin model type which should be denoted by {@link DataPrepperPlugin#pluginConfigurationType()}
+ */
 class PluginConfigurationConverter {
     private final ObjectMapper objectMapper;
+    private final Validator validator;
 
     PluginConfigurationConverter() {
         this.objectMapper = new ObjectMapper().setPropertyNamingStrategy(PropertyNamingStrategies.SNAKE_CASE);
+
+        final ValidatorFactory validationFactory = Validation.buildDefaultValidatorFactory();
+        validator = validationFactory.getValidator();
     }
 
+    /**
+     * Converts and validates to a plugin model type. The conversion happens via
+     * Java Bean Validation 2.0.
+     *
+     * @param pluginConfigurationType the destination type
+     * @param pluginSetting The source {@link PluginSetting}
+     * @return The converted object of type pluginConfigurationType
+     * @throws InvalidPluginConfigurationException - If the plugin configuration is invalid
+     */
     public Object convert(final Class<?> pluginConfigurationType, final PluginSetting pluginSetting) {
         Objects.requireNonNull(pluginConfigurationType);
         Objects.requireNonNull(pluginSetting);
 
         if(pluginConfigurationType.equals(PluginSetting.class))
             return pluginSetting;
 
-        return objectMapper.convertValue(pluginSetting.getSettings(), pluginConfigurationType);
+        final Object configuration = objectMapper.convertValue(pluginSetting.getSettings(), pluginConfigurationType);
+
+        final Set<ConstraintViolation<Object>> constraintViolations = validator.validate(configuration);
+
+        if(!constraintViolations.isEmpty()) {
+            final String violationsString = constraintViolations.stream()
+                    .map(v -> v.getPropertyPath().toString() + " " + v.getMessage())
+                    .collect(Collectors.joining(". "));
+
+            final String exceptionMessage = String.format("Plugin %s in pipeline %s is configured incorrectly: %s",
+                    pluginSetting.getName(), pluginSetting.getPipelineName(), violationsString);
+            throw new InvalidPluginConfigurationException(exceptionMessage);
+        }
+
+        return configuration;
     }
 }

data-prepper-core/src/test/java/com/amazon/dataprepper/plugin/DefaultPluginFactoryIT.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package com.amazon.dataprepper.plugin;
+
+import com.amazon.dataprepper.model.configuration.PluginSetting;
+import com.amazon.dataprepper.model.plugin.InvalidPluginConfigurationException;
+import com.amazon.dataprepper.plugins.TestPlugin;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.UUID;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/**
+ * Integration test of the plugin framework. These tests should not mock any portion
+ * of the plugin framework. But, they may mock inputs when appropriate.
+ */
+class DefaultPluginFactoryIT {
+
+    private String pluginName;
+    private String pipelineName;
+
+    @BeforeEach
+    void setUp() {
+        pluginName = "test_plugin";
+        pipelineName = UUID.randomUUID().toString();
+    }
+
+    private DefaultPluginFactory createObjectUnderTest() {
+        return new DefaultPluginFactory();
+    }
+
+    @Test
+    void loadPlugin_should_return_a_new_plugin_instance_with_the_expected_configuration() {
+
+        final String requiredStringValue = UUID.randomUUID().toString();
+        final String optionalStringValue = UUID.randomUUID().toString();
+
+        final Map<String, Object> pluginSettingMap = new HashMap<>();
+        pluginSettingMap.put("required_string", requiredStringValue);
+        pluginSettingMap.put("optional_string", optionalStringValue);
+        final PluginSetting pluginSetting = createPluginSettings(pluginSettingMap);
+
+        final TestPluggableInterface plugin = createObjectUnderTest().loadPlugin(TestPluggableInterface.class, pluginSetting);
+
+        assertThat(plugin, instanceOf(TestPlugin.class));
+
+        final TestPlugin testPlugin = (TestPlugin) plugin;
+
+        final TestPluginConfiguration configuration = testPlugin.getConfiguration();
+
+        assertThat(configuration.getRequiredString(), equalTo(requiredStringValue));
+        assertThat(configuration.getOptionalString(), equalTo(optionalStringValue));
+    }
+
+    @Test
+    void loadPlugin_should_throw_when_a_plugin_configuration_is_invalid() {
+        final String optionalStringValue = UUID.randomUUID().toString();
+
+        final Map<String, Object> pluginSettingMap = new HashMap<>();
+        pluginSettingMap.put("optional_string", optionalStringValue);
+        final PluginSetting pluginSetting = createPluginSettings(pluginSettingMap);
+
+        final DefaultPluginFactory objectUnderTest = createObjectUnderTest();
+
+        final InvalidPluginConfigurationException actualException = assertThrows(InvalidPluginConfigurationException.class,
+                () -> objectUnderTest.loadPlugin(TestPluggableInterface.class, pluginSetting));
+
+        assertThat(actualException.getMessage(), notNullValue());
+        assertThat(actualException.getMessage(), equalTo("Plugin test_plugin in pipeline " + pipelineName + " is configured incorrectly: requiredString must not be null"));
+    }
+
+    private PluginSetting createPluginSettings(final Map<String, Object> pluginSettingMap) {
+        final PluginSetting pluginSetting = new PluginSetting(pluginName, pluginSettingMap);
+        pluginSetting.setPipelineName(pipelineName);
+        return pluginSetting;
+    }
+}

data-prepper-core/src/test/java/com/amazon/dataprepper/plugin/PluginConfigurationConverterTest.java

@@ -6,26 +6,38 @@
 package com.amazon.dataprepper.plugin;
 
 import com.amazon.dataprepper.model.configuration.PluginSetting;
+import com.amazon.dataprepper.model.plugin.InvalidPluginConfigurationException;
+import jakarta.validation.ConstraintViolation;
+import jakarta.validation.Path;
+import jakarta.validation.Validation;
+import jakarta.validation.Validator;
+import jakarta.validation.ValidatorFactory;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
 
 import java.util.Collections;
 import java.util.UUID;
 
+import static org.hamcrest.CoreMatchers.containsString;
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.instanceOf;
 import static org.hamcrest.CoreMatchers.notNullValue;
 import static org.hamcrest.CoreMatchers.sameInstance;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.then;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
 
 class PluginConfigurationConverterTest {
     private PluginSetting pluginSetting;
+    private Validator validator;
 
     static class TestConfiguration {
+        @SuppressWarnings("unused")
         private String myValue;
 
         public String getMyValue() {
@@ -36,10 +48,19 @@ public String getMyValue() {
     @BeforeEach
     void setUp() {
         pluginSetting = mock(PluginSetting.class);
+
+        validator = mock(Validator.class);
     }
 
     private PluginConfigurationConverter createObjectUnderTest() {
-        return new PluginConfigurationConverter();
+        final ValidatorFactory validatorFactory = mock(ValidatorFactory.class);
+        given(validatorFactory.getValidator()).willReturn(validator);
+
+        try(final MockedStatic<Validation> validationMockedStatic = mockStatic(Validation.class)) {
+            validationMockedStatic.when(Validation::buildDefaultValidatorFactory)
+                    .thenReturn(validatorFactory);
+            return new PluginConfigurationConverter();
+        }
     }
 
     @Test
@@ -64,6 +85,7 @@ void convert_with_PluginSetting_target_should_return_pluginSetting_object_direct
                 sameInstance(pluginSetting));
 
         then(pluginSetting).shouldHaveNoInteractions();
+        then(validator).shouldHaveNoInteractions();
     }
 
     @Test
@@ -83,4 +105,54 @@ void convert_with_other_target_should_return_pluginSetting_object_directly() {
         assertThat(convertedTestConfiguration.getMyValue(), equalTo(value));
     }
 
+    @Test
+    void convert_with_other_target_should_validate_configuration() {
+
+        final String value = UUID.randomUUID().toString();
+        given(pluginSetting.getSettings())
+                .willReturn(Collections.singletonMap("my_value", value));
+
+        final Object convertedConfiguration = createObjectUnderTest().convert(TestConfiguration.class, pluginSetting);
+
+        then(validator)
+                .should()
+                .validate(convertedConfiguration);
+    }
+
+    @Test
+    void convert_with_other_target_should_throw_exception_when_there_are_constraint_violations() {
+
+        final String value = UUID.randomUUID().toString();
+        given(pluginSetting.getSettings())
+                .willReturn(Collections.singletonMap("my_value", value));
+
+        final String pluginName = UUID.randomUUID().toString();
+        given(pluginSetting.getName())
+                .willReturn(pluginName);
+
+        final String pipelineName = UUID.randomUUID().toString();
+        given(pluginSetting.getPipelineName())
+                .willReturn(pipelineName);
+
+        @SuppressWarnings("unchecked") final ConstraintViolation<Object> constraintViolation = mock(ConstraintViolation.class);
+        final String errorMessage = UUID.randomUUID().toString();
+        given(constraintViolation.getMessage()).willReturn(errorMessage);
+        final String propertyPathString = UUID.randomUUID().toString();
+        final Path propertyPath = mock(Path.class);
+        given(propertyPath.toString()).willReturn(propertyPathString);
+        given(constraintViolation.getPropertyPath()).willReturn(propertyPath);
+
+        given(validator.validate(any()))
+                .willReturn(Collections.singleton(constraintViolation));
+
+        final PluginConfigurationConverter objectUnderTest = createObjectUnderTest();
+
+        final InvalidPluginConfigurationException actualException = assertThrows(InvalidPluginConfigurationException.class,
+                () -> objectUnderTest.convert(TestConfiguration.class, pluginSetting));
+
+        assertThat(actualException.getMessage(), containsString(pluginName));
+        assertThat(actualException.getMessage(), containsString(pipelineName));
+        assertThat(actualException.getMessage(), containsString(propertyPathString));
+        assertThat(actualException.getMessage(), containsString(errorMessage));
+    }
 }

data-prepper-core/src/test/java/com/amazon/dataprepper/plugin/TestPluggableInterface.java

@@ -0,0 +1,9 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package com.amazon.dataprepper.plugin;
+
+public interface TestPluggableInterface {
+}

data-prepper-core/src/test/java/com/amazon/dataprepper/plugin/TestPluginConfiguration.java

@@ -0,0 +1,31 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package com.amazon.dataprepper.plugin;
+
+import jakarta.validation.constraints.NotNull;
+
+public class TestPluginConfiguration {
+    @NotNull
+    private String requiredString;
+
+    private String optionalString;
+
+    public String getRequiredString() {
+        return requiredString;
+    }
+
+    public void setRequiredString(final String requiredString) {
+        this.requiredString = requiredString;
+    }
+
+    public String getOptionalString() {
+        return optionalString;
+    }
+
+    public void setOptionalString(final String optionalString) {
+        this.optionalString = optionalString;
+    }
+}