Merge branch 'opensearch-project:main' into remote-detector-calls

engechas · Feb 10, 2024 · 4a20f75 · 4a20f75
2 parents 3a88a0c + 9c4dd59
commit 4a20f75
Show file tree

Hide file tree

Showing 81 changed files with 2,218 additions and 502 deletions.
diff --git a/.github/workflows/kafka-plugin-integration-tests.yml b/.github/workflows/kafka-plugin-integration-tests.yml
@@ -43,7 +43,7 @@ jobs:
           ./gradlew data-prepper-plugins:kafka-plugins:integrationTest -Dtests.kafka.bootstrap_servers=localhost:9092 -Dtests.kafka.authconfig.username=admin -Dtests.kafka.authconfig.password=admin --tests KafkaStartIT
       - name: Run Kafka integration tests
         run: |
-          ./gradlew data-prepper-plugins:kafka-plugins:integrationTest -Dtests.kafka.bootstrap_servers=localhost:9092 -Dtests.kafka.authconfig.username=admin -Dtests.kafka.authconfig.password=admin --tests KafkaSourceJsonTypeIT --tests '*kafka.buffer*'
+          ./gradlew data-prepper-plugins:kafka-plugins:integrationTest -Dtests.kafka.bootstrap_servers=localhost:9092 -Dtests.kafka.authconfig.username=admin -Dtests.kafka.authconfig.password=admin --tests KafkaSourceJsonTypeIT --tests KafkaBufferIT --tests KafkaBufferOTelIT
 
       - name: Upload Unit Test Results
         if: always()

diff --git a/data-prepper-core/src/main/java/org/opensearch/dataprepper/plugin/DefaultPluginFactory.java b/data-prepper-core/src/main/java/org/opensearch/dataprepper/plugin/DefaultPluginFactory.java
@@ -44,7 +44,7 @@ public class DefaultPluginFactory implements PluginFactory {
     @Inject
     DefaultPluginFactory(
             final PluginProviderLoader pluginProviderLoader,
-            final PluginCreator pluginCreator,
+            @Named("pluginCreator") final PluginCreator pluginCreator,
             final PluginConfigurationConverter pluginConfigurationConverter,
             final PluginBeanFactoryProvider pluginBeanFactoryProvider,
             final PluginConfigurationObservableFactory pluginConfigurationObservableFactory,
@@ -113,7 +113,7 @@ private <T> ComponentPluginArgumentsContext getConstructionContext(final PluginS
         final Class<?> pluginConfigurationType = pluginAnnotation.pluginConfigurationType();
         final Object configuration = pluginConfigurationConverter.convert(pluginConfigurationType, pluginSetting);
         final PluginConfigObservable pluginConfigObservable = pluginConfigurationObservableFactory
-                .createDefaultPluginConfigObservable(pluginConfigurationConverter, pluginClass, pluginSetting);
+                .createDefaultPluginConfigObservable(pluginConfigurationConverter, pluginConfigurationType, pluginSetting);
 
         return new ComponentPluginArgumentsContext.Builder()
                 .withPluginSetting(pluginSetting)

diff --git a/data-prepper-core/src/main/java/org/opensearch/dataprepper/plugin/ExtensionLoader.java b/data-prepper-core/src/main/java/org/opensearch/dataprepper/plugin/ExtensionLoader.java
@@ -20,24 +20,25 @@
 public class ExtensionLoader {
     private final ExtensionPluginConfigurationConverter extensionPluginConfigurationConverter;
     private final ExtensionClassProvider extensionClassProvider;
-    private final PluginCreator pluginCreator;
+    private final PluginCreator extensionPluginCreator;
 
     @Inject
     ExtensionLoader(
             final ExtensionPluginConfigurationConverter extensionPluginConfigurationConverter,
             final ExtensionClassProvider extensionClassProvider,
-            final PluginCreator pluginCreator) {
+            @Named("extensionPluginCreator") final PluginCreator extensionPluginCreator) {
         this.extensionPluginConfigurationConverter = extensionPluginConfigurationConverter;
         this.extensionClassProvider = extensionClassProvider;
-        this.pluginCreator = pluginCreator;
+        this.extensionPluginCreator = extensionPluginCreator;
     }
 
     List<? extends ExtensionPlugin> loadExtensions() {
         return extensionClassProvider.loadExtensionPluginClasses()
                 .stream()
                 .map(extensionClass -> {
                     final PluginArgumentsContext pluginArgumentsContext = getConstructionContext(extensionClass);
-                    return pluginCreator.newPluginInstance(extensionClass, pluginArgumentsContext, convertClassToName(extensionClass));
+                    return extensionPluginCreator.newPluginInstance(
+                            extensionClass, pluginArgumentsContext, convertClassToName(extensionClass));
                 })
                 .collect(Collectors.toList());
     }

diff --git a/...rc/main/java/org/opensearch/dataprepper/plugin/PluginConfigurationObservableRegister.java b/...rc/main/java/org/opensearch/dataprepper/plugin/PluginConfigurationObservableRegister.java
@@ -7,6 +7,7 @@
 
 import org.opensearch.dataprepper.model.plugin.PluginConfigObservable;
 import org.opensearch.dataprepper.model.plugin.PluginConfigPublisher;
+import org.springframework.context.annotation.DependsOn;
 
 import javax.inject.Inject;
 import javax.inject.Named;
@@ -15,6 +16,7 @@
 import java.util.Set;
 
 @Named
+@DependsOn("extensionsApplier")
 public class PluginConfigurationObservableRegister {
     private final Set<PluginConfigPublisher> pluginConfigPublishers;
 

diff --git a/data-prepper-core/src/main/java/org/opensearch/dataprepper/plugin/PluginCreator.java b/data-prepper-core/src/main/java/org/opensearch/dataprepper/plugin/PluginCreator.java
@@ -12,8 +12,6 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import javax.inject.Inject;
-import javax.inject.Named;
 import java.lang.reflect.Constructor;
 import java.lang.reflect.InvocationTargetException;
 import java.util.Arrays;
@@ -22,13 +20,15 @@
 import java.util.Optional;
 import java.util.stream.Collectors;
 
-@Named
 class PluginCreator {
     private static final Logger LOG = LoggerFactory.getLogger(PluginCreator.class);
 
     private final PluginConfigurationObservableRegister pluginConfigurationObservableRegister;
 
-    @Inject
+    PluginCreator() {
+        this.pluginConfigurationObservableRegister = null;
+    }
+
     PluginCreator(final PluginConfigurationObservableRegister pluginConfigurationObservableRegister) {
         this.pluginConfigurationObservableRegister = pluginConfigurationObservableRegister;
     }
@@ -45,7 +45,9 @@ <T> T newPluginInstance(final Class<T> pluginClass,
 
         final Object[] constructorArguments = pluginArgumentsContext.createArguments(constructor.getParameterTypes(), args);
 
-        pluginConfigurationObservableRegister.registerPluginConfigurationObservables(constructorArguments);
+        if (pluginConfigurationObservableRegister != null) {
+            pluginConfigurationObservableRegister.registerPluginConfigurationObservables(constructorArguments);
+        }
 
         try {
             return (T) constructor.newInstance(constructorArguments);

diff --git a/data-prepper-core/src/main/java/org/opensearch/dataprepper/plugin/PluginCreatorContext.java b/data-prepper-core/src/main/java/org/opensearch/dataprepper/plugin/PluginCreatorContext.java
@@ -0,0 +1,19 @@
+package org.opensearch.dataprepper.plugin;
+
+import org.springframework.context.annotation.Bean;
+
+import javax.inject.Named;
+
+@Named
+public class PluginCreatorContext {
+    @Bean(name = "extensionPluginCreator")
+    public PluginCreator observablePluginCreator() {
+        return new PluginCreator();
+    }
+
+    @Bean(name = "pluginCreator")
+    public PluginCreator pluginCreator(
+            final PluginConfigurationObservableRegister pluginConfigurationObservableRegister) {
+        return new PluginCreator(pluginConfigurationObservableRegister);
+    }
+}
diff --git a/...repper-core/src/test/java/org/opensearch/dataprepper/plugin/DefaultPluginFactoryTest.java b/...repper-core/src/test/java/org/opensearch/dataprepper/plugin/DefaultPluginFactoryTest.java
@@ -203,6 +203,8 @@ void loadPlugin_should_create_a_new_instance_of_the_first_plugin_found() {
 
             assertThat(createObjectUnderTest().loadPlugin(baseClass, pluginSetting),
                     equalTo(expectedInstance));
+            verify(pluginConfigurationObservableFactory).createDefaultPluginConfigObservable(eq(pluginConfigurationConverter),
+                    eq(PluginSetting.class), eq(pluginSetting));
             verify(beanFactoryProvider).get();
         }
 
@@ -254,6 +256,8 @@ void loadPlugins_should_return_a_single_instance_when_the_the_numberOfInstances_
                     baseClass, pluginSetting, c -> 1);
 
             verify(beanFactoryProvider).get();
+            verify(pluginConfigurationObservableFactory).createDefaultPluginConfigObservable(eq(pluginConfigurationConverter),
+                    eq(PluginSetting.class), eq(pluginSetting));
             final ArgumentCaptor<ComponentPluginArgumentsContext> pluginArgumentsContextArgCapture = ArgumentCaptor.forClass(ComponentPluginArgumentsContext.class);
             verify(pluginCreator).newPluginInstance(eq(expectedPluginClass), pluginArgumentsContextArgCapture.capture(), eq(pluginName));
             final ComponentPluginArgumentsContext actualPluginArgumentsContext = pluginArgumentsContextArgCapture.getValue();
@@ -281,6 +285,8 @@ void loadPlugin_with_varargs_should_return_a_single_instance_when_the_the_number
             final Object plugin = createObjectUnderTest().loadPlugin(baseClass, pluginSetting, object);
 
             verify(beanFactoryProvider).get();
+            verify(pluginConfigurationObservableFactory).createDefaultPluginConfigObservable(eq(pluginConfigurationConverter),
+                    eq(PluginSetting.class), eq(pluginSetting));
             final ArgumentCaptor<ComponentPluginArgumentsContext> pluginArgumentsContextArgCapture = ArgumentCaptor.forClass(ComponentPluginArgumentsContext.class);
             verify(pluginCreator).newPluginInstance(eq(expectedPluginClass), pluginArgumentsContextArgCapture.capture(), eq(pluginName), eq(object));
             final ComponentPluginArgumentsContext actualPluginArgumentsContext = pluginArgumentsContextArgCapture.getValue();

diff --git a/data-prepper-core/src/test/java/org/opensearch/dataprepper/plugin/ExtensionLoaderTest.java b/data-prepper-core/src/test/java/org/opensearch/dataprepper/plugin/ExtensionLoaderTest.java
@@ -49,12 +49,12 @@ class ExtensionLoaderTest {
     @Mock
     private ExtensionClassProvider extensionClassProvider;
     @Mock
-    private PluginCreator pluginCreator;
+    private PluginCreator extensionPluginCreator;
     @Captor
     private ArgumentCaptor<PluginArgumentsContext> pluginArgumentsContextArgumentCaptor;
 
     private ExtensionLoader createObjectUnderTest() {
-        return new ExtensionLoader(extensionPluginConfigurationConverter, extensionClassProvider, pluginCreator);
+        return new ExtensionLoader(extensionPluginConfigurationConverter, extensionClassProvider, extensionPluginCreator);
     }
 
     @Test
@@ -74,7 +74,7 @@ void loadExtensions_returns_single_extension_for_single_plugin_class() {
         when(extensionClassProvider.loadExtensionPluginClasses()).thenReturn(Collections.singleton(pluginClass));
 
         final ExtensionPlugin expectedPlugin = mock(ExtensionPlugin.class);
-        when(pluginCreator.newPluginInstance(
+        when(extensionPluginCreator.newPluginInstance(
                 eq(pluginClass),
                 any(PluginArgumentsContext.class),
                 startsWith("extension_plugin")))
@@ -98,15 +98,15 @@ void loadExtensions_returns_single_extension_with_config_for_single_plugin_class
         final String expectedPluginName = "test_extension_with_config";
         when(extensionPluginConfigurationConverter.convert(eq(true), eq(TestExtensionConfig.class),
                 eq("/test_extension"))).thenReturn(testExtensionConfig);
-        when(pluginCreator.newPluginInstance(
+        when(extensionPluginCreator.newPluginInstance(
                 eq(TestExtensionWithConfig.class),
                 any(PluginArgumentsContext.class),
                 eq(expectedPluginName)))
                 .thenReturn(expectedPlugin);
 
         final List<? extends ExtensionPlugin> extensionPlugins = createObjectUnderTest().loadExtensions();
 
-        verify(pluginCreator).newPluginInstance(eq(TestExtensionWithConfig.class),
+        verify(extensionPluginCreator).newPluginInstance(eq(TestExtensionWithConfig.class),
                 pluginArgumentsContextArgumentCaptor.capture(), eq(expectedPluginName));
         assertThat(pluginArgumentsContextArgumentCaptor.getValue(), instanceOf(
                 ExtensionLoader.SingleConfigArgumentArgumentsContext.class));
@@ -128,7 +128,7 @@ void loadExtensions_returns_multiple_extensions_for_multiple_plugin_classes() {
             final String expectedPluginName = ExtensionLoader.classNameToPluginName(pluginClass.getSimpleName());
             final ExtensionPlugin extensionPlugin = mock((Class<ExtensionPlugin>)pluginClass);
 
-            when(pluginCreator.newPluginInstance(
+            when(extensionPluginCreator.newPluginInstance(
                     eq(pluginClass),
                     any(PluginArgumentsContext.class),
                     eq(expectedPluginName)))
@@ -156,7 +156,7 @@ void loadExtensions_invokes_newPluginInstance_with_PluginArgumentsContext_not_su
 
         when(extensionClassProvider.loadExtensionPluginClasses()).thenReturn(Collections.singleton(pluginClass));
 
-        when(pluginCreator.newPluginInstance(
+        when(extensionPluginCreator.newPluginInstance(
                 any(Class.class),
                 any(PluginArgumentsContext.class),
                 anyString()))
@@ -166,7 +166,7 @@ void loadExtensions_invokes_newPluginInstance_with_PluginArgumentsContext_not_su
 
         final ArgumentCaptor<PluginArgumentsContext> contextArgumentCaptor =
                 ArgumentCaptor.forClass(PluginArgumentsContext.class);
-        verify(pluginCreator).newPluginInstance(
+        verify(extensionPluginCreator).newPluginInstance(
                 eq(pluginClass),
                 contextArgumentCaptor.capture(),
                 anyString());
@@ -183,7 +183,7 @@ void loadExtensions_invokes_newPluginInstance_with_PluginArgumentsContext_which_
 
         when(extensionClassProvider.loadExtensionPluginClasses()).thenReturn(Collections.singleton(pluginClass));
 
-        when(pluginCreator.newPluginInstance(
+        when(extensionPluginCreator.newPluginInstance(
                 any(Class.class),
                 any(PluginArgumentsContext.class),
                 anyString()))
@@ -193,7 +193,7 @@ void loadExtensions_invokes_newPluginInstance_with_PluginArgumentsContext_which_
 
         final ArgumentCaptor<PluginArgumentsContext> contextArgumentCaptor =
                 ArgumentCaptor.forClass(PluginArgumentsContext.class);
-        verify(pluginCreator).newPluginInstance(
+        verify(extensionPluginCreator).newPluginInstance(
                 eq(pluginClass),
                 contextArgumentCaptor.capture(),
                 any());