Merge branch 'main' into parse-json-depth

Signed-off-by: Krishna Kondaka <[email protected]>

build.gradle

@@ -245,6 +245,10 @@ subprojects {
         javaLauncher = javaToolchains.launcherFor {
             languageVersion = JavaLanguageVersion.current()
         }
+        testLogging {
+            exceptionFormat "full"
+            showStackTraces false
+        }
         reports {
             junitXml.required
             html.required

data-prepper-api/build.gradle

@@ -13,8 +13,8 @@ dependencies {
     implementation 'com.fasterxml.jackson.datatype:jackson-datatype-jsr310'
     implementation 'com.fasterxml.jackson.datatype:jackson-datatype-jdk8'
     implementation libs.parquet.common
-    testImplementation 'com.fasterxml.jackson.dataformat:jackson-dataformat-yaml'
     implementation libs.commons.lang3
+    testImplementation 'com.fasterxml.jackson.dataformat:jackson-dataformat-yaml'
     testImplementation project(':data-prepper-test-common')
     testImplementation 'org.skyscreamer:jsonassert:1.5.3'
     testImplementation libs.commons.io

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/annotations/DataPrepperPlugin.java

@@ -72,4 +72,16 @@
      * @since 1.2
      */
     Class<?> pluginConfigurationType() default PluginSetting.class;
+
+    /**
+     * Optional Packages to scan for Data Prepper DI components.
+     * Plugins provide this list if they want to use Dependency Injection in its module.
+     * Providing this value, implicitly assumes and initiates plugin specific isolated ApplicationContext.
+     * <p>
+     * The package names that spring context scans will be picked up by these marker classes.
+     *
+     * @return Array of classes to use for package scan.
+     * @since 2.2
+     */
+    Class[] packagesToScan() default {};
 }

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/annotations/UsesDataPrepperPlugin.java

@@ -0,0 +1,23 @@
+package org.opensearch.dataprepper.model.annotations;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Annotates a field that uses Data Prepper plugin config as its value.
+ */
+@Documented
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.FIELD})
+public @interface UsesDataPrepperPlugin {
+    /**
+     * The class type for this plugin.
+     *
+     * @return The Java class
+     * @since 1.2
+     */
+    Class<?> pluginType();
+}

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/configuration/ConditionalRoute.java

@@ -31,7 +31,7 @@
  */
 @JsonPropertyOrder
 @JsonClassDescription("The key-value pair defines routing condition, where the key is the name of a route and the " +
-        "value is a Data Prepper expression representing the routing condition.")
+        "value is an expression representing the routing condition.")
 @JsonSerialize(using = ConditionalRoute.ConditionalRouteSerializer.class)
 @JsonDeserialize(using = ConditionalRoute.ConditionalRouteDeserializer.class)
 public class ConditionalRoute {

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/event/DataType.java

@@ -95,7 +95,7 @@ public String getTypeName() {
     }
 
     @JsonCreator
-    static DataType fromTypeName(final String option) {
+    public static DataType fromTypeName(final String option) {
         return TYPES_MAP.get(option);
     }
 

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/event/HandleFailedEventsOption.java

@@ -6,6 +6,7 @@
 package org.opensearch.dataprepper.model.event;
 
 import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonValue;
 
 import java.util.Arrays;
 import java.util.Map;
@@ -45,4 +46,9 @@ public boolean shouldLog() {
     static HandleFailedEventsOption fromOptionValue(final String option) {
         return OPTIONS_MAP.get(option.toLowerCase());
     }
+
+    @JsonValue
+    public String toOptionValue() {
+        return option;
+    }
 }

data-prepper-api/src/test/java/org/opensearch/dataprepper/model/event/DataTypeTest.java

@@ -6,14 +6,21 @@
 package org.opensearch.dataprepper.model.event;
 
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtensionContext;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.ArgumentsProvider;
+import org.junit.jupiter.params.provider.ArgumentsSource;
 import org.junit.jupiter.params.provider.MethodSource;
 import org.junit.jupiter.params.provider.EnumSource;
 
 import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.CoreMatchers.not;
+import static org.hamcrest.CoreMatchers.notNullValue;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.emptyString;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
 
 import java.math.BigDecimal;
 import java.util.ArrayList;
@@ -39,26 +46,55 @@ void test_isSameType(Object object, String type, boolean expectedResult) {
         assertThat(DataType.isSameType(object, type), equalTo(expectedResult));
     }
 
+    @ParameterizedTest
+    @EnumSource(DataType.class)
+    void getTypeName_returns_non_empty_string_for_all_types(final DataType dataType) {
+        assertThat(dataType.getTypeName(), notNullValue());
+        assertThat(dataType.getTypeName(), not(emptyString()));
+    }
+
+    @ParameterizedTest
+    @ArgumentsSource(DataTypeToKnownString.class)
+    void getTypeName_returns_expected_name(final DataType dataType, final String expectedString) {
+        assertThat(dataType.getTypeName(), equalTo(expectedString));
+    }
+
     private static Stream<Arguments> getSameTypeTestData() {
         int[] testArray = {1,2};
         List<Integer> testList = new ArrayList<>();
         return Stream.of(
-            Arguments.of(2, "integer", true),
-            Arguments.of("testString", "string", true),
-            Arguments.of(2L, "long", true),
-            Arguments.of(2.0, "double", true),
-            Arguments.of(BigDecimal.valueOf(2.34567), "big_decimal", true),
-            Arguments.of(true, "boolean", true),
-            Arguments.of(Map.of("k","v"), "map", true),
-            Arguments.of(testArray, "array", true),
-            Arguments.of(testList, "array", true),
-            Arguments.of(2.0, "integer", false),
-            Arguments.of(2, "string", false),
-            Arguments.of("testString", "long", false),
-            Arguments.of("testString", "double", false),
-            Arguments.of(2, "boolean", false),
-            Arguments.of(2L, "map", false),
-            Arguments.of(2, "array", false)
+            arguments(2, "integer", true),
+            arguments("testString", "string", true),
+            arguments(2L, "long", true),
+            arguments(2.0, "double", true),
+            arguments(BigDecimal.valueOf(2.34567), "big_decimal", true),
+            arguments(true, "boolean", true),
+            arguments(Map.of("k","v"), "map", true),
+            arguments(testArray, "array", true),
+            arguments(testList, "array", true),
+            arguments(2.0, "integer", false),
+            arguments(2, "string", false),
+            arguments("testString", "long", false),
+            arguments("testString", "double", false),
+            arguments(2, "boolean", false),
+            arguments(2L, "map", false),
+            arguments(2, "array", false)
         );
     }
+
+    static class DataTypeToKnownString implements ArgumentsProvider {
+        @Override
+        public Stream<? extends Arguments> provideArguments(final ExtensionContext extensionContext) {
+            return Stream.of(
+                    arguments(DataType.STRING, "string"),
+                    arguments(DataType.BOOLEAN, "boolean"),
+                    arguments(DataType.INTEGER, "integer"),
+                    arguments(DataType.LONG, "long"),
+                    arguments(DataType.DOUBLE, "double"),
+                    arguments(DataType.BIG_DECIMAL, "big_decimal"),
+                    arguments(DataType.MAP, "map"),
+                    arguments(DataType.ARRAY, "array")
+            );
+        }
+    }
 }

data-prepper-api/src/test/java/org/opensearch/dataprepper/model/event/HandleFailedEventsOptionTest.java

@@ -5,29 +5,84 @@
 
 package org.opensearch.dataprepper.model.event;
 
-import org.hamcrest.CoreMatchers;
+import org.junit.jupiter.api.extension.ExtensionContext;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.ArgumentsProvider;
+import org.junit.jupiter.params.provider.ArgumentsSource;
 import org.junit.jupiter.params.provider.EnumSource;
 
+import java.util.stream.Stream;
+
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.notNullValue;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
 
 class HandleFailedEventsOptionTest {
+    @ParameterizedTest
+    @ArgumentsSource(EnumToShouldLogArgumentsProvider.class)
+    void shouldLog_returns_expected_value(final HandleFailedEventsOption option, final boolean shouldLog) {
+        assertThat(option.shouldLog(), equalTo(shouldLog));
+    }
+
+    @ParameterizedTest
+    @ArgumentsSource(EnumToShouldShouldDropArgumentsProvider.class)
+    void shouldDropEvent_returns_expected_value(final HandleFailedEventsOption option, final boolean shouldDrop) {
+        assertThat(option.shouldDropEvent(), equalTo(shouldDrop));
+    }
+
+    @ParameterizedTest
+    @ArgumentsSource(EnumToOptionValueArgumentsProvider.class)
+    void toOptionValue_returns_expected_value(final HandleFailedEventsOption option, final String optionValue) {
+        assertThat(option.toOptionValue(), equalTo(optionValue));
+    }
+
+    @ParameterizedTest
+    @ArgumentsSource(EnumToOptionValueArgumentsProvider.class)
+    void fromOptionValue_returns_expected_option(final HandleFailedEventsOption option, final String optionValue) {
+        assertThat(HandleFailedEventsOption.fromOptionValue(optionValue), equalTo(option));
+    }
+
     @ParameterizedTest
     @EnumSource(HandleFailedEventsOption.class)
-    void fromOptionValue(final HandleFailedEventsOption option) {
-        assertThat(HandleFailedEventsOption.fromOptionValue(option.name()), CoreMatchers.is(option));
+    void toOptionValue_returns_non_null_for_all(final HandleFailedEventsOption option) {
+        assertThat(option.toOptionValue(), notNullValue());
+    }
 
-        if (option == HandleFailedEventsOption.SKIP || option == HandleFailedEventsOption.SKIP_SILENTLY) {
-            assertThat(option.shouldDropEvent(), equalTo(false));
-        } else {
-            assertThat(option.shouldDropEvent(), equalTo(true));
+    private static class EnumToOptionValueArgumentsProvider implements ArgumentsProvider {
+        @Override
+        public Stream<? extends Arguments> provideArguments(final ExtensionContext context) {
+            return Stream.of(
+                    arguments(HandleFailedEventsOption.SKIP, "skip"),
+                    arguments(HandleFailedEventsOption.SKIP_SILENTLY, "skip_silently"),
+                    arguments(HandleFailedEventsOption.DROP, "drop"),
+                    arguments(HandleFailedEventsOption.DROP_SILENTLY, "drop_silently")
+            );
         }
+    }
+
+    private static class EnumToShouldShouldDropArgumentsProvider implements ArgumentsProvider {
+        @Override
+        public Stream<? extends Arguments> provideArguments(final ExtensionContext context) {
+            return Stream.of(
+                    arguments(HandleFailedEventsOption.SKIP, false),
+                    arguments(HandleFailedEventsOption.SKIP_SILENTLY, false),
+                    arguments(HandleFailedEventsOption.DROP, true),
+                    arguments(HandleFailedEventsOption.DROP_SILENTLY, true)
+            );
+        }
+    }
 
-        if (option == HandleFailedEventsOption.SKIP_SILENTLY || option == HandleFailedEventsOption.DROP_SILENTLY) {
-            assertThat(option.shouldLog(), equalTo(false));
-        } else {
-            assertThat(option.shouldLog(), equalTo(true));
+    private static class EnumToShouldLogArgumentsProvider implements ArgumentsProvider {
+        @Override
+        public Stream<? extends Arguments> provideArguments(final ExtensionContext context) {
+            return Stream.of(
+                    arguments(HandleFailedEventsOption.SKIP, true),
+                    arguments(HandleFailedEventsOption.DROP, true),
+                    arguments(HandleFailedEventsOption.SKIP_SILENTLY, false),
+                    arguments(HandleFailedEventsOption.DROP_SILENTLY, false)
+            );
         }
     }
 }

data-prepper-core/src/integrationTest/java/org/opensearch/dataprepper/plugin/DefaultPluginFactoryIT.java

@@ -15,7 +15,10 @@
 import org.opensearch.dataprepper.model.configuration.PipelinesDataFlowModel;
 import org.opensearch.dataprepper.model.configuration.PluginSetting;
 import org.opensearch.dataprepper.model.plugin.InvalidPluginConfigurationException;
+import org.opensearch.dataprepper.model.source.Source;
 import org.opensearch.dataprepper.plugins.TestObjectPlugin;
+import org.opensearch.dataprepper.plugins.test.TestComponent;
+import org.opensearch.dataprepper.plugins.test.TestDISource;
 import org.opensearch.dataprepper.plugins.test.TestPlugin;
 import org.opensearch.dataprepper.validation.LoggingPluginErrorsHandler;
 import org.opensearch.dataprepper.validation.PluginErrorCollector;
@@ -30,6 +33,8 @@
 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.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
 /**
@@ -96,6 +101,23 @@ void loadPlugin_should_return_a_new_plugin_instance_with_the_expected_configurat
         assertThat(configuration.getOptionalString(), equalTo(optionalStringValue));
     }
 
+    @Test
+    void loadPlugin_should_return_a_new_plugin_instance_with_DI_context_initialized() {
+
+        final Map<String, Object> pluginSettingMap = new HashMap<>();
+        final PluginSetting pluginSetting = new PluginSetting("test_di_source", pluginSettingMap);
+        pluginSetting.setPipelineName(pipelineName);
+
+        final Source sourcePlugin = createObjectUnderTest().loadPlugin(Source.class, pluginSetting);
+
+        assertThat(sourcePlugin, instanceOf(TestDISource.class));
+        TestDISource plugin = (TestDISource) sourcePlugin;
+        // Testing the auto wired been with the Dependency Injection
+        assertNotNull(plugin.getTestComponent());
+        assertInstanceOf(TestComponent.class, plugin.getTestComponent());
+        assertThat(plugin.getTestComponent().getIdentifier(), equalTo("test-component"));
+    }
+
     @Test
     void loadPlugin_should_return_a_new_plugin_instance_with_the_expected_configuration_variable_args() {
 

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

@@ -11,6 +11,8 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonSetter;
 import com.fasterxml.jackson.annotation.Nulls;
+import org.opensearch.dataprepper.core.event.EventConfiguration;
+import org.opensearch.dataprepper.core.event.EventConfigurationContainer;
 import org.opensearch.dataprepper.model.configuration.PipelineExtensions;
 import org.opensearch.dataprepper.model.configuration.PluginModel;
 import org.opensearch.dataprepper.parser.config.MetricTagFilter;
@@ -29,7 +31,7 @@
 /**
  * Class to hold configuration for DataPrepper, including server port and Log4j settings
  */
-public class DataPrepperConfiguration implements ExtensionsConfiguration {
+public class DataPrepperConfiguration implements ExtensionsConfiguration, EventConfigurationContainer {
     static final Duration DEFAULT_SHUTDOWN_DURATION = Duration.ofSeconds(30L);
 
     private static final String DEFAULT_SOURCE_COORDINATION_STORE = "in_memory";
@@ -47,6 +49,7 @@ public class DataPrepperConfiguration implements ExtensionsConfiguration {
     private CircuitBreakerConfig circuitBreakerConfig;
     private SourceCoordinationConfig sourceCoordinationConfig;
     private PipelineShutdownOption pipelineShutdown;
+    private EventConfiguration eventConfiguration;
     private Map<String, String> metricTags = new HashMap<>();
     private List<MetricTagFilter> metricTagFilters = new LinkedList<>();
     private PeerForwarderConfiguration peerForwarderConfiguration;
@@ -92,6 +95,7 @@ public DataPrepperConfiguration(
             @JsonProperty("circuit_breakers") final CircuitBreakerConfig circuitBreakerConfig,
             @JsonProperty("source_coordination") final SourceCoordinationConfig sourceCoordinationConfig,
             @JsonProperty("pipeline_shutdown") final PipelineShutdownOption pipelineShutdown,
+            @JsonProperty("event") final EventConfiguration eventConfiguration,
             @JsonProperty("extensions")
             @JsonInclude(JsonInclude.Include.NON_NULL)
             @JsonSetter(nulls = Nulls.SKIP)
@@ -102,6 +106,7 @@ public DataPrepperConfiguration(
                 ? new SourceCoordinationConfig(new PluginModel(DEFAULT_SOURCE_COORDINATION_STORE, Collections.emptyMap()), null)
                 : sourceCoordinationConfig;
         this.pipelineShutdown = pipelineShutdown != null ? pipelineShutdown : DEFAULT_PIPELINE_SHUTDOWN;
+        this.eventConfiguration = eventConfiguration != null ? eventConfiguration : EventConfiguration.defaultConfiguration();
         setSsl(ssl);
         this.keyStoreFilePath = keyStoreFilePath != null ? keyStoreFilePath : "";
         this.keyStorePassword = keyStorePassword != null ? keyStorePassword : "";
@@ -226,6 +231,10 @@ public PipelineShutdownOption getPipelineShutdown() {
         return pipelineShutdown;
     }
 
+    public EventConfiguration getEventConfiguration() {
+        return eventConfiguration;
+    }
+
     @Override
     public PipelineExtensions getPipelineExtensions() {
         return pipelineExtensions;

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

@@ -131,6 +131,7 @@ void setUp() {
 
     @AfterEach
     void tearDown() {
+        verify(dataPrepperConfiguration).getEventConfiguration();
         verifyNoMoreInteractions(dataPrepperConfiguration);
     }