feat(osgi): add opt-in Apache Felix plugin framework

Add an opt-in plugin-loading mode backed by an embedded Apache Felix OSGi
framework, enabled with the system property -Dplugin.framework=osgi. The
default classpath loader remains the default when the flag is not set. When
the flag is not set, plugin discovery is functionally identical to today: the
same ServiceLoader-based providers are returned in the same order, the OSGi
framework never starts, and no OSGi code path executes. The only added work in
legacy mode is the construction of a couple of inert Spring beans.

The new plugin-framework-osgi module embeds the Felix lifecycle
(FelixPluginManager), bootstraps it in OSGi mode through a Spring-managed
runner (OsgiFrameworkRunner), and exposes a PluginProvider backed by the OSGi
service registry (OsgiPluginRegistry). At startup in OSGi mode,
StaticBundleLoader installs each legacy plugin JAR through BundleAdapter, which
installs already-OSGi JARs directly and wraps legacy (non-OSGi) JARs by
generating a versioned OSGi manifest. Wrapped plugins' @DataPrepperPlugin
classes are registered as OSGi services (LegacyPluginBundleActivator). Data
Prepper API packages are exported to bundles with semver version attributes
read from data-prepper-api build metadata (DataPrepperOsgiPackages,
DataPrepperApiVersion), consistent with the data-prepper-api
backward-compatibility contract from #6607, so a plugin built on 2.15 resolves
on a 2.16 host. DataPrepperOsgiPackages also exports the framework package
org.opensearch.dataprepper.plugin.osgi so adapted bundles can load the
LegacyPluginBundleActivator from the system bundle.

StaticBundleLoader installs, resolves, and starts bundles with fail-fast
behavior, a startup summary log, and Micrometer metrics.
BundleResolutionErrorTranslator turns Felix resolution failures into readable
diagnostics, and BundleHealthCheck verifies framework, bundle, and classloader
isolation state. BundleClassLoaderScope manages the thread context classloader
at the plugin invocation boundary so ServiceLoader and SPI resolve against the
bundle classloader. PluginHealthProbe is a seam for future functional health
checks.

The OsgiFrameworkRunner bean is always instantiated by Spring but does nothing
unless -Dplugin.framework=osgi is set, because it returns early in its
@PostConstruct method. The Felix framework JAR ships on the runtime classpath
as a small (<1MB) library with zero transitive dependencies, but it is not
exercised in legacy mode.

Scope is limited to classloader isolation and deploy-time validation. Bundles
install once at startup and stop once at shutdown, the same lifecycle as
classpath plugins today. No production hot-reload is included. The
PluginHotLoader helper is test-scoped only and not on the production classpath.

Wiring changes: settings.gradle adds the module, a Felix version-catalog
entry, and the bnd plugin; build-resources.gradle adds the module to
coreProjects; data-prepper-core declares a runtimeOnly dependency on the new
module; PluginProviderLoader is made public with a registerProvider() hook and
caches the framework-mode flag at construction; a test in
data-prepper-plugin-framework's PluginProviderLoaderTest verifies the
mode-caching behavior. Legacy behavior is unchanged. A new, permanent, minimal
example plugin data-prepper-plugins/echo-processor is added to the build to
demonstrate a minimal @DataPrepperPlugin and the SPI path.

Resolves #6760

Signed-off-by: Kiran Kumar Veeravelly <veeravkk@amazon.com>

Author: Kiran Kumar Veeravelly

build-resources.gradle

@@ -19,5 +19,6 @@ ext.coreProjects = [
         project(':data-prepper-test'),
         project(':data-prepper-plugin-framework'),
         project(':data-prepper-plugin-schema'),
-        project(':data-prepper-plugin-schema-cli')
+        project(':data-prepper-plugin-schema-cli'),
+        project(':plugin-framework-osgi')
 ]

data-prepper-core/build.gradle

@@ -37,6 +37,7 @@ dependencies {
     implementation project(':data-prepper-logstash-configuration')
     implementation project(':data-prepper-pipeline-parser')
     implementation project(':data-prepper-plugin-framework')
+    runtimeOnly project(':plugin-framework-osgi')
     testImplementation project(':data-prepper-plugin-framework').sourceSets.test.output
     testImplementation project(':data-prepper-plugins:common').sourceSets.test.output
     testImplementation project(':data-prepper-plugins:file-source')

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

@@ -1,6 +1,11 @@
 /*
  * 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;
@@ -9,24 +14,56 @@
 import org.slf4j.LoggerFactory;
 
 import javax.inject.Named;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.List;
 import java.util.ServiceLoader;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.stream.Collectors;
 import java.util.stream.StreamSupport;
 
+/**
+ * Loads and provides plugin providers via SPI and optional runtime registration.
+ * Made public solely for the OSGi integration module to register additional providers.
+ */
 @Named
-class PluginProviderLoader {
+public class PluginProviderLoader {
     private static final Logger LOG = LoggerFactory.getLogger(PluginProviderLoader.class);
-    private final ServiceLoader<PluginProvider> serviceLoader;
+    private static final String PLUGIN_FRAMEWORK_PROPERTY = "plugin.framework";
+    private static final String MODE_OSGI = "osgi";
+    private final List<PluginProvider> classpathProviders;
+    private final List<PluginProvider> additionalProviders = new CopyOnWriteArrayList<>();
+    private final String frameworkMode;
 
     PluginProviderLoader() {
-        serviceLoader = ServiceLoader.load(PluginProvider.class);
+        final ServiceLoader<PluginProvider> serviceLoader = ServiceLoader.load(PluginProvider.class);
+        classpathProviders = StreamSupport.stream(serviceLoader.spliterator(), false)
+                .collect(Collectors.toList());
+        frameworkMode = System.getProperty(PLUGIN_FRAMEWORK_PROPERTY, "legacy");
+    }
+
+    /**
+     * Registers an additional PluginProvider at runtime.
+     * Used by the OSGi integration to inject the OsgiPluginRegistry.
+     *
+     * @param provider the provider to add
+     */
+    public void registerProvider(final PluginProvider provider) {
+        additionalProviders.add(provider);
+        LOG.info("Registered additional PluginProvider: {}", provider.getClass().getSimpleName());
     }
 
     Collection<PluginProvider> getPluginProviders() {
-        final List<PluginProvider> pluginProviders = StreamSupport.stream(serviceLoader.spliterator(), false)
-                .collect(Collectors.toList());
+        final List<PluginProvider> pluginProviders;
+
+        if (MODE_OSGI.equalsIgnoreCase(frameworkMode) && !additionalProviders.isEmpty()) {
+            pluginProviders = new ArrayList<>(additionalProviders);
+            pluginProviders.addAll(classpathProviders);
+            LOG.debug("Plugin framework running in OSGi mode with {} providers ({} OSGi + {} classpath)",
+                    pluginProviders.size(), additionalProviders.size(), classpathProviders.size());
+        } else {
+            pluginProviders = new ArrayList<>(classpathProviders);
+        }
 
         LOG.debug("Data Prepper is configured with {} distinct plugin providers.", pluginProviders.size());
 

data-prepper-plugin-framework/src/test/java/org/opensearch/dataprepper/plugin/PluginProviderLoaderTest.java

@@ -1,6 +1,11 @@
 /*
  * 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;
@@ -66,4 +71,28 @@ void getPluginProviders_returns_a_collection_of_all_PluginProvider_instances() {
         assertThat(actualPluginProviders, not(sameInstance(originalPluginProviders)));
         assertThat(actualPluginProviders, equalTo(originalPluginProviders));
     }
+
+    @Test
+    void getPluginProviders_caches_mode_at_construction_and_ignores_later_property_change() {
+        // Construct with legacy mode (default)
+        given(serviceLoader.spliterator())
+                .willReturn(Collections.<PluginProvider>emptyList().spliterator());
+        final PluginProviderLoader loader = createObjectUnderTest();
+
+        // Register an additional provider
+        final PluginProvider osgiProvider = mock(PluginProvider.class);
+        loader.registerProvider(osgiProvider);
+
+        // Even if we change the system property now, the mode was cached at construction
+        System.setProperty("plugin.framework", "osgi");
+        try {
+            given(serviceLoader.spliterator())
+                    .willReturn(Collections.<PluginProvider>emptyList().spliterator());
+            final Collection<PluginProvider> providers = loader.getPluginProviders();
+            // Should use legacy behavior since the mode was cached as "legacy" at construction
+            assertThat(providers.size(), equalTo(0));
+        } finally {
+            System.clearProperty("plugin.framework");
+        }
+    }
 }

data-prepper-plugins/echo-processor/build.gradle

@@ -0,0 +1,16 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+plugins {
+    id 'java'
+}
+
+dependencies {
+    implementation project(':data-prepper-api')
+}
+
+test {
+    useJUnitPlatform()
+}

data-prepper-plugins/echo-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/echo/EchoProcessor.java

@@ -0,0 +1,58 @@
+/*
+ * 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.processor.echo;
+
+import org.opensearch.dataprepper.metrics.PluginMetrics;
+import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
+import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
+import org.opensearch.dataprepper.model.event.Event;
+import org.opensearch.dataprepper.model.processor.AbstractProcessor;
+import org.opensearch.dataprepper.model.processor.Processor;
+import org.opensearch.dataprepper.model.record.Record;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+
+/**
+ * A simple pass-through processor that logs each event and returns them unchanged.
+ * Used to validate the OSGi plugin loading pipeline end-to-end.
+ */
+@DataPrepperPlugin(name = "echo", pluginType = Processor.class)
+public class EchoProcessor extends AbstractProcessor<Record<Event>, Record<Event>> {
+    private static final Logger LOG = LoggerFactory.getLogger(EchoProcessor.class);
+
+    @DataPrepperPluginConstructor
+    public EchoProcessor(final PluginMetrics pluginMetrics) {
+        super(pluginMetrics);
+    }
+
+    @Override
+    public Collection<Record<Event>> doExecute(final Collection<Record<Event>> records) {
+        for (final Record<Event> record : records) {
+            LOG.debug("echo: {}", record.getData().toJsonString());
+        }
+        return records;
+    }
+
+    @Override
+    public void prepareForShutdown() {
+    }
+
+    @Override
+    public boolean isReadyForShutdown() {
+        return true;
+    }
+
+    @Override
+    public void shutdown() {
+    }
+}

data-prepper-plugins/echo-processor/src/main/resources/META-INF/data-prepper.plugins.properties

@@ -0,0 +1,6 @@
+#
+# Copyright OpenSearch Contributors
+# SPDX-License-Identifier: Apache-2.0
+#
+
+org.opensearch.dataprepper.plugin.packages=org.opensearch.dataprepper.plugins

data-prepper-plugins/echo-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/echo/EchoProcessorTest.java

@@ -0,0 +1,76 @@
+/*
+ * 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.processor.echo;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.opensearch.dataprepper.metrics.PluginMetrics;
+import org.opensearch.dataprepper.model.event.Event;
+import org.opensearch.dataprepper.model.event.JacksonEvent;
+import org.opensearch.dataprepper.model.record.Record;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+@ExtendWith(MockitoExtension.class)
+class EchoProcessorTest {
+
+    @Mock
+    private PluginMetrics pluginMetrics;
+
+    private EchoProcessor echoProcessor;
+
+    @BeforeEach
+    void setUp() {
+        echoProcessor = new EchoProcessor(pluginMetrics);
+    }
+
+    @Test
+    void doExecute_returns_same_records() {
+        final Record<Event> record = new Record<>(JacksonEvent.builder()
+                .withEventType("event")
+                .withData(Collections.singletonMap("key", "value"))
+                .build());
+
+        final Collection<Record<Event>> result = echoProcessor.doExecute(Arrays.asList(record));
+
+        assertThat(result.size(), is(1));
+        assertThat(result.iterator().next(), is(record));
+    }
+
+    @Test
+    void doExecute_with_empty_collection_returns_empty() {
+        final Collection<Record<Event>> result = echoProcessor.doExecute(Collections.emptyList());
+        assertThat(result.isEmpty(), is(true));
+    }
+
+    @Test
+    void isReadyForShutdown_returns_true() {
+        assertThat(echoProcessor.isReadyForShutdown(), is(true));
+    }
+
+    @Test
+    void shutdown_does_not_throw() {
+        echoProcessor.shutdown();
+    }
+
+    @Test
+    void prepareForShutdown_does_not_throw() {
+        echoProcessor.prepareForShutdown();
+    }
+}