Support automatic plugin loading in Data Prepper core.

This change allows plugin authors to load other plugins by defining @UsesDataPrepperPlugin on a field with a different type. It will automatically load the plugin using the PluginFactory and give the main plugin the desired plugin instance that they are looking for.

This updates the sqs source to use this new capability for its InputCodec defined by the codec property.

Resolves #4838

Signed-off-by: David Venable <dlv@amazon.com>

Author: dlvenable

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

@@ -26,7 +26,10 @@
 import org.opensearch.dataprepper.model.plugin.NoPluginFoundException;
 import org.opensearch.dataprepper.model.source.Source;
 import org.opensearch.dataprepper.pipeline.parser.DataPrepperDeserializationProblemHandler;
+import org.opensearch.dataprepper.plugins.TestNestedPluginInterface;
 import org.opensearch.dataprepper.plugins.TestObjectPlugin;
+import org.opensearch.dataprepper.plugins.TestPluginWithNestedPlugin;
+import org.opensearch.dataprepper.plugins.TestPluginWithNestedPluginConfig;
 import org.opensearch.dataprepper.plugins.configtest.TestComponentWithConfigInject;
 import org.opensearch.dataprepper.plugins.configtest.TestDISourceWithConfig;
 import org.opensearch.dataprepper.plugins.test.TestComponent;
@@ -250,6 +253,34 @@ void loadPlugin_should_succeed_when_a_non_experimental_plugin_has_an_experimenta
         assertThat(plugin, notNullValue());
     }
 
+    @Test
+    void loadPlugin_should_resolve_nested_plugin_annotated_with_UsesDataPrepperPlugin() {
+        final String nameValue = UUID.randomUUID().toString();
+        final String nestedTestValue = UUID.randomUUID().toString();
+
+        final Map<String, Object> nestedPluginSettings = new HashMap<>();
+        nestedPluginSettings.put("test_value", nestedTestValue);
+
+        final Map<String, Object> pluginSettingMap = new HashMap<>();
+        pluginSettingMap.put("name", nameValue);
+        pluginSettingMap.put("nested_plugin", Collections.singletonMap("test_nested_plugin", nestedPluginSettings));
+
+        final PluginSetting pluginSetting = new PluginSetting("test_plugin_with_nested", pluginSettingMap);
+        pluginSetting.setPipelineName(pipelineName);
+
+        final TestPluggableInterface plugin = createObjectUnderTest().loadPlugin(TestPluggableInterface.class, pluginSetting);
+
+        assertThat(plugin, instanceOf(TestPluginWithNestedPlugin.class));
+
+        final TestPluginWithNestedPlugin testPlugin = (TestPluginWithNestedPlugin) plugin;
+        final TestPluginWithNestedPluginConfig configuration = testPlugin.getConfiguration();
+
+        assertThat(configuration.getName(), equalTo(nameValue));
+        assertThat(configuration.getNestedPlugin(), notNullValue());
+        assertThat(configuration.getNestedPlugin(), instanceOf(TestNestedPluginInterface.class));
+        assertThat(configuration.getNestedPlugin().getValue(), equalTo(nestedTestValue));
+    }
+
     private PluginSetting createPluginSettings(final Map<String, Object> pluginSettingMap) {
         final PluginSetting pluginSetting = new PluginSetting(pluginName, pluginSettingMap);
         pluginSetting.setPipelineName(pipelineName);

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

@@ -0,0 +1,28 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.dataprepper.plugins;
+
+import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
+import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
+
+@DataPrepperPlugin(name = "test_nested_plugin", pluginType = TestNestedPluginInterface.class, pluginConfigurationType = TestNestedPluginConfig.class)
+public class TestNestedPlugin implements TestNestedPluginInterface {
+    private final TestNestedPluginConfig config;
+
+    @DataPrepperPluginConstructor
+    public TestNestedPlugin(final TestNestedPluginConfig config) {
+        this.config = config;
+    }
+
+    @Override
+    public String getValue() {
+        return config.getTestValue();
+    }
+}

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

@@ -0,0 +1,21 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.dataprepper.plugins;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+
+public class TestNestedPluginConfig {
+    @JsonProperty("test_value")
+    private String testValue;
+
+    public String getTestValue() {
+        return testValue;
+    }
+}

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

@@ -0,0 +1,14 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.dataprepper.plugins;
+
+public interface TestNestedPluginInterface {
+    String getValue();
+}

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

@@ -0,0 +1,28 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.dataprepper.plugins;
+
+import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
+import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
+import org.opensearch.dataprepper.plugin.TestPluggableInterface;
+
+@DataPrepperPlugin(name = "test_plugin_with_nested", pluginType = TestPluggableInterface.class, pluginConfigurationType = TestPluginWithNestedPluginConfig.class)
+public class TestPluginWithNestedPlugin implements TestPluggableInterface {
+    private final TestPluginWithNestedPluginConfig configuration;
+
+    @DataPrepperPluginConstructor
+    public TestPluginWithNestedPlugin(final TestPluginWithNestedPluginConfig configuration) {
+        this.configuration = configuration;
+    }
+
+    public TestPluginWithNestedPluginConfig getConfiguration() {
+        return configuration;
+    }
+}

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

@@ -0,0 +1,30 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.dataprepper.plugins;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.opensearch.dataprepper.model.annotations.UsesDataPrepperPlugin;
+
+public class TestPluginWithNestedPluginConfig {
+    @JsonProperty("nested_plugin")
+    @UsesDataPrepperPlugin(pluginType = TestNestedPluginInterface.class)
+    private TestNestedPluginInterface nestedPlugin;
+
+    @JsonProperty("name")
+    private String name;
+
+    public TestNestedPluginInterface getNestedPlugin() {
+        return nestedPlugin;
+    }
+
+    public String getName() {
+        return name;
+    }
+}

data-prepper-plugin-framework/src/main/java/org/opensearch/dataprepper/plugin/DataPrepperPluginBeanDeserializerModifier.java

@@ -0,0 +1,82 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.dataprepper.plugin;
+
+import com.fasterxml.jackson.databind.BeanDescription;
+import com.fasterxml.jackson.databind.DeserializationConfig;
+import com.fasterxml.jackson.databind.deser.BeanDeserializerModifier;
+import com.fasterxml.jackson.databind.deser.BeanDeserializerBuilder;
+import com.fasterxml.jackson.databind.deser.SettableBeanProperty;
+import com.fasterxml.jackson.databind.introspect.AnnotatedField;
+import com.fasterxml.jackson.databind.introspect.AnnotatedMember;
+import com.fasterxml.jackson.databind.introspect.BeanPropertyDefinition;
+import org.opensearch.dataprepper.model.annotations.UsesDataPrepperPlugin;
+import org.opensearch.dataprepper.model.configuration.PluginModel;
+
+import java.util.Iterator;
+import java.util.List;
+
+class DataPrepperPluginBeanDeserializerModifier extends BeanDeserializerModifier {
+
+    @Override
+    public BeanDeserializerBuilder updateBuilder(
+            final DeserializationConfig config,
+            final BeanDescription beanDesc,
+            final BeanDeserializerBuilder builder) {
+
+        final List<BeanPropertyDefinition> properties = beanDesc.findProperties();
+
+        for (final BeanPropertyDefinition propertyDef : properties) {
+            final UsesDataPrepperPlugin annotation = findAnnotation(propertyDef);
+            if (annotation == null) {
+                continue;
+            }
+
+            if (PluginModel.class.isAssignableFrom(propertyDef.getRawPrimaryType())) {
+                continue;
+            }
+
+            final Class<?> pluginType = annotation.pluginType();
+            final NestedPluginDeserializer deserializer = new NestedPluginDeserializer(pluginType);
+
+            final Iterator<SettableBeanProperty> propertyIterator = builder.getProperties();
+            while (propertyIterator.hasNext()) {
+                final SettableBeanProperty property = propertyIterator.next();
+                if (property.getName().equals(propertyDef.getName())) {
+                    final SettableBeanProperty updatedProperty = property.withValueDeserializer(deserializer);
+                    builder.addOrReplaceProperty(updatedProperty, true);
+                    break;
+                }
+            }
+        }
+
+        return builder;
+    }
+
+    private UsesDataPrepperPlugin findAnnotation(final BeanPropertyDefinition propertyDef) {
+        final AnnotatedField field = propertyDef.getField();
+        if (field != null) {
+            final UsesDataPrepperPlugin annotation = field.getAnnotation(UsesDataPrepperPlugin.class);
+            if (annotation != null) {
+                return annotation;
+            }
+        }
+
+        final AnnotatedMember mutator = propertyDef.getMutator();
+        if (mutator != null) {
+            final UsesDataPrepperPlugin annotation = mutator.getAnnotation(UsesDataPrepperPlugin.class);
+            if (annotation != null) {
+                return annotation;
+            }
+        }
+
+        return null;
+    }
+}

data-prepper-plugin-framework/src/main/java/org/opensearch/dataprepper/plugin/DefaultPluginConfigObservable.java

@@ -1,30 +1,38 @@
 /*
  * Copyright OpenSearch Contributors
  * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
  */
 
 package org.opensearch.dataprepper.plugin;
 
 import org.opensearch.dataprepper.model.configuration.PluginSetting;
 import org.opensearch.dataprepper.model.plugin.PluginConfigObservable;
 import org.opensearch.dataprepper.model.plugin.PluginConfigObserver;
+import org.opensearch.dataprepper.model.plugin.PluginFactory;
 
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
-public class DefaultPluginConfigObservable implements PluginConfigObservable {
+class DefaultPluginConfigObservable implements PluginConfigObservable {
     private final Map<PluginConfigObserver, Boolean> pluginConfigObserverBooleanMap
             = new ConcurrentHashMap<>();
     private final PluginConfigurationConverter pluginConfigurationConverter;
     private final Class<?> pluginConfigClass;
     private final PluginSetting rawPluginSettings;
+    private final PluginFactory pluginFactory;
 
-    public DefaultPluginConfigObservable(final PluginConfigurationConverter pluginConfigurationConverter,
-                                         final Class<?> pluginConfigClass,
-                                         final PluginSetting rawPluginSettings) {
+    DefaultPluginConfigObservable(final PluginConfigurationConverter pluginConfigurationConverter,
+                                  final Class<?> pluginConfigClass,
+                                  final PluginSetting rawPluginSettings,
+                                  final PluginFactory pluginFactory) {
         this.pluginConfigurationConverter = pluginConfigurationConverter;
         this.pluginConfigClass = pluginConfigClass;
         this.rawPluginSettings = rawPluginSettings;
+        this.pluginFactory = pluginFactory;
     }
 
     @Override
@@ -36,7 +44,7 @@ public boolean addPluginConfigObserver(final PluginConfigObserver pluginConfigOb
     @Override
     public void update() {
         final Object newPluginConfiguration = pluginConfigurationConverter.convert(
-                pluginConfigClass, rawPluginSettings);
+                pluginConfigClass, rawPluginSettings, pluginFactory);
         pluginConfigObserverBooleanMap.keySet().forEach(
                 pluginConfigObserver -> pluginConfigObserver.update(newPluginConfiguration));
     }

data-prepper-plugin-framework/src/main/java/org/opensearch/dataprepper/plugin/DefaultPluginFactory.java

@@ -1,6 +1,10 @@
 /*
  * Copyright OpenSearch Contributors
  * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
  */
 
 package org.opensearch.dataprepper.plugin;
@@ -117,9 +121,9 @@ private <T> ComponentPluginArgumentsContext getConstructionContext(final PluginS
 
         final Class<?> pluginConfigurationType = pluginAnnotation.pluginConfigurationType();
 
-        final Object configuration = pluginConfigurationConverter.convert(pluginConfigurationType, pluginSetting);
+        final Object configuration = pluginConfigurationConverter.convert(pluginConfigurationType, pluginSetting, this);
         final PluginConfigObservable pluginConfigObservable = pluginConfigurationObservableFactory
-                .createDefaultPluginConfigObservable(pluginConfigurationConverter, pluginConfigurationType, pluginSetting);
+                .createDefaultPluginConfigObservable(pluginConfigurationConverter, pluginConfigurationType, pluginSetting, this);
 
         Class[] markersToScan = pluginAnnotation.packagesToScan();
         BeanFactory beanFactory = pluginBeanFactoryProvider.createPluginSpecificContext(markersToScan, configuration, pluginSetting);