Introducing core package in the data-prepper-core module (opensearch-…

…project#5056) core package introduced in the data-prepper-core module Signed-off-by: Santhosh Gandhe <[email protected]>
san81 · Oct 15, 2024 · 3e220d4 · 3e220d4
1 parent 472b246
commit 3e220d4
Show file tree

Hide file tree

Showing 244 changed files with 852 additions and 829 deletions.
diff --git a/...re/src/integrationTest/java/org/opensearch/dataprepper/integration/MinimalPipelineIT.java b/...re/src/integrationTest/java/org/opensearch/dataprepper/integration/MinimalPipelineIT.java
@@ -8,12 +8,12 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import org.opensearch.dataprepper.test.framework.DataPrepperTestRunner;
 import org.opensearch.dataprepper.model.event.Event;
 import org.opensearch.dataprepper.model.event.JacksonEvent;
 import org.opensearch.dataprepper.model.record.Record;
 import org.opensearch.dataprepper.plugins.InMemorySinkAccessor;
 import org.opensearch.dataprepper.plugins.InMemorySourceAccessor;
+import org.opensearch.dataprepper.test.framework.DataPrepperTestRunner;
 
 import java.util.Collections;
 import java.util.List;

diff --git a/.../src/integrationTest/java/org/opensearch/dataprepper/integration/PipelinesWithAcksIT.java b/.../src/integrationTest/java/org/opensearch/dataprepper/integration/PipelinesWithAcksIT.java
@@ -7,26 +7,26 @@
 
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
-import org.opensearch.dataprepper.test.framework.DataPrepperTestRunner;
 import org.opensearch.dataprepper.model.event.Event;
 import org.opensearch.dataprepper.model.record.Record;
 import org.opensearch.dataprepper.plugins.InMemorySinkAccessor;
 import org.opensearch.dataprepper.plugins.InMemorySourceAccessor;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.Assert.assertFalse;
+import org.opensearch.dataprepper.test.framework.DataPrepperTestRunner;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.time.Instant;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
-import java.time.Instant;
 
 import static org.awaitility.Awaitility.await;
 import static org.hamcrest.CoreMatchers.equalTo;
-import static org.hamcrest.Matchers.lessThanOrEqualTo;
 import static org.hamcrest.CoreMatchers.not;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.empty;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+import static org.junit.Assert.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 class PipelinesWithAcksIT {
     private static final Logger LOG = LoggerFactory.getLogger(PipelinesWithAcksIT.class);

diff --git a/...re/src/integrationTest/java/org/opensearch/dataprepper/plugin/DefaultPluginFactoryIT.java b/...re/src/integrationTest/java/org/opensearch/dataprepper/plugin/DefaultPluginFactoryIT.java
@@ -10,8 +10,11 @@
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
-import org.opensearch.dataprepper.acknowledgements.DefaultAcknowledgementSetManager;
+import org.opensearch.dataprepper.core.acknowledgements.DefaultAcknowledgementSetManager;
 import org.opensearch.dataprepper.core.event.EventFactoryApplicationContextMarker;
+import org.opensearch.dataprepper.core.validation.LoggingPluginErrorsHandler;
+import org.opensearch.dataprepper.core.validation.PluginErrorCollector;
+import org.opensearch.dataprepper.validation.PluginErrorsHandler;
 import org.opensearch.dataprepper.model.configuration.PipelinesDataFlowModel;
 import org.opensearch.dataprepper.model.configuration.PluginSetting;
 import org.opensearch.dataprepper.model.plugin.InvalidPluginConfigurationException;
@@ -20,9 +23,6 @@
 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;
-import org.opensearch.dataprepper.validation.PluginErrorsHandler;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 
 import java.util.HashMap;

diff --git a/...repper-core/src/integrationTest/java/org/opensearch/dataprepper/plugins/ExtensionsIT.java b/...repper-core/src/integrationTest/java/org/opensearch/dataprepper/plugins/ExtensionsIT.java
@@ -14,26 +14,27 @@
 import org.junit.jupiter.api.extension.ExtendWith;
 import org.mockito.Mock;
 import org.mockito.junit.jupiter.MockitoExtension;
-import org.opensearch.dataprepper.acknowledgements.DefaultAcknowledgementSetManager;
-import org.opensearch.dataprepper.breaker.CircuitBreakerManager;
+import org.opensearch.dataprepper.core.acknowledgements.DefaultAcknowledgementSetManager;
+import org.opensearch.dataprepper.core.breaker.CircuitBreakerManager;
 import org.opensearch.dataprepper.core.event.EventFactoryApplicationContextMarker;
+import org.opensearch.dataprepper.core.parser.config.DataPrepperAppConfiguration;
+import org.opensearch.dataprepper.core.parser.config.FileStructurePathProvider;
+import org.opensearch.dataprepper.core.parser.config.PipelineParserConfiguration;
+import org.opensearch.dataprepper.core.peerforwarder.PeerForwarderProvider;
+import org.opensearch.dataprepper.core.pipeline.router.RouterFactory;
+import org.opensearch.dataprepper.core.sourcecoordination.SourceCoordinatorFactory;
+import org.opensearch.dataprepper.core.validation.LoggingPluginErrorsHandler;
+import org.opensearch.dataprepper.core.validation.PluginErrorCollector;
+import org.opensearch.dataprepper.expression.StartsWithExpressionFunction;
+import org.opensearch.dataprepper.validation.PluginErrorsHandler;
 import org.opensearch.dataprepper.model.acknowledgements.AcknowledgementSetManager;
 import org.opensearch.dataprepper.model.configuration.PluginSetting;
 import org.opensearch.dataprepper.model.event.EventFactory;
 import org.opensearch.dataprepper.model.plugin.PluginFactory;
-import org.opensearch.dataprepper.parser.config.DataPrepperAppConfiguration;
-import org.opensearch.dataprepper.parser.config.FileStructurePathProvider;
-import org.opensearch.dataprepper.parser.config.PipelineParserConfiguration;
-import org.opensearch.dataprepper.peerforwarder.PeerForwarderProvider;
-import org.opensearch.dataprepper.pipeline.router.RouterFactory;
 import org.opensearch.dataprepper.plugin.DefaultPluginFactory;
 import org.opensearch.dataprepper.plugin.ObjectMapperConfiguration;
 import org.opensearch.dataprepper.plugin.TestPluggableInterface;
 import org.opensearch.dataprepper.plugins.test.TestExtension;
-import org.opensearch.dataprepper.sourcecoordination.SourceCoordinatorFactory;
-import org.opensearch.dataprepper.validation.LoggingPluginErrorsHandler;
-import org.opensearch.dataprepper.validation.PluginErrorCollector;
-import org.opensearch.dataprepper.validation.PluginErrorsHandler;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 
 import java.util.ArrayList;
@@ -92,6 +93,7 @@ void setUp() {
         coreContext.scan(DefaultAcknowledgementSetManager.class.getPackage().getName());
 
         coreContext.scan(DefaultPluginFactory.class.getPackage().getName());
+        coreContext.scan(StartsWithExpressionFunction.class.getPackage().getName());
 
         when(fileStructurePathProvider.getPipelineConfigFileLocation()).thenReturn(
                 "src/test/resources/valid_pipeline.yml"

diff --git a/...repper-core/src/integrationTest/java/org/opensearch/dataprepper/plugins/InMemorySink.java b/...repper-core/src/integrationTest/java/org/opensearch/dataprepper/plugins/InMemorySink.java
@@ -5,13 +5,13 @@
 
 package org.opensearch.dataprepper.plugins;
 
+import org.opensearch.dataprepper.model.acknowledgements.AcknowledgementSetManager;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
-import org.opensearch.dataprepper.model.acknowledgements.AcknowledgementSetManager;
 import org.opensearch.dataprepper.model.event.Event;
+import org.opensearch.dataprepper.model.event.EventHandle;
 import org.opensearch.dataprepper.model.record.Record;
 import org.opensearch.dataprepper.model.sink.Sink;
-import org.opensearch.dataprepper.model.event.EventHandle;
 
 import java.util.Collection;
 

diff --git a/...pper-core/src/integrationTest/java/org/opensearch/dataprepper/plugins/InMemorySource.java b/...pper-core/src/integrationTest/java/org/opensearch/dataprepper/plugins/InMemorySource.java
@@ -5,21 +5,21 @@
 
 package org.opensearch.dataprepper.plugins;
 
+import org.opensearch.dataprepper.model.acknowledgements.AcknowledgementSet;
+import org.opensearch.dataprepper.model.acknowledgements.AcknowledgementSetManager;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
 import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
-import org.opensearch.dataprepper.model.acknowledgements.AcknowledgementSetManager;
-import org.opensearch.dataprepper.model.acknowledgements.AcknowledgementSet;
-import org.opensearch.dataprepper.model.event.EventFactory;
 import org.opensearch.dataprepper.model.buffer.Buffer;
 import org.opensearch.dataprepper.model.event.Event;
+import org.opensearch.dataprepper.model.event.EventFactory;
 import org.opensearch.dataprepper.model.record.Record;
 import org.opensearch.dataprepper.model.source.Source;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.time.Duration;
 
 /**
  * A Data Prepper source which can receive records via the {@link InMemorySourceAccessor}.

diff --git a/...integrationTest/java/org/opensearch/dataprepper/test/framework/DataPrepperTestRunner.java b/...integrationTest/java/org/opensearch/dataprepper/test/framework/DataPrepperTestRunner.java
@@ -6,8 +6,8 @@
 package org.opensearch.dataprepper.test.framework;
 
 import org.opensearch.dataprepper.AbstractContextManager;
-import org.opensearch.dataprepper.DataPrepper;
-import org.opensearch.dataprepper.parser.config.FileStructurePathProvider;
+import org.opensearch.dataprepper.core.DataPrepper;
+import org.opensearch.dataprepper.core.parser.config.FileStructurePathProvider;
 import org.opensearch.dataprepper.plugins.InMemorySinkAccessor;
 import org.opensearch.dataprepper.plugins.InMemorySourceAccessor;
 import org.slf4j.Logger;

diff --git a/data-prepper-core/src/main/java/org/opensearch/dataprepper/AbstractContextManager.java b/data-prepper-core/src/main/java/org/opensearch/dataprepper/AbstractContextManager.java
@@ -5,6 +5,7 @@
 
 package org.opensearch.dataprepper;
 
+import org.opensearch.dataprepper.core.DataPrepper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
@@ -24,7 +25,11 @@
 public abstract class AbstractContextManager {
     private static final Logger LOG = LoggerFactory.getLogger(AbstractContextManager.class);
     private static final String BASE_DATA_PREPPER_PACKAGE = "org.opensearch.dataprepper";
-    private static final String EXPRESSION_PACKAGE = BASE_DATA_PREPPER_PACKAGE + ".expression";
+    private static final String[] BASE_DATA_PREPPER_PACKAGES = {
+            BASE_DATA_PREPPER_PACKAGE + ".core",
+            BASE_DATA_PREPPER_PACKAGE + ".plugin"
+    };
+    private static final String EXPRESSION_PACKAGE =   BASE_DATA_PREPPER_PACKAGE + ".expression";
 
     private final AnnotationConfigApplicationContext publicApplicationContext;
     private final AnnotationConfigApplicationContext coreApplicationContext;
@@ -53,7 +58,7 @@ private void start() {
 
         publicApplicationContext.refresh();
         coreApplicationContext.setParent(publicApplicationContext);
-        coreApplicationContext.scan(BASE_DATA_PREPPER_PACKAGE);
+        coreApplicationContext.scan(BASE_DATA_PREPPER_PACKAGES);
         preRefreshCoreApplicationContext(coreApplicationContext);
 
         coreApplicationContext.refresh();

diff --git a/...g/opensearch/dataprepper/DataPrepper.java → ...nsearch/dataprepper/core/DataPrepper.java b/...g/opensearch/dataprepper/DataPrepper.java → ...nsearch/dataprepper/core/DataPrepper.java
@@ -3,17 +3,19 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package org.opensearch.dataprepper;
+package org.opensearch.dataprepper.core;
 
 import io.micrometer.core.instrument.util.StringUtils;
+import org.opensearch.dataprepper.DataPrepperShutdownListener;
+import org.opensearch.dataprepper.DataPrepperShutdownOptions;
+import org.opensearch.dataprepper.core.parser.PipelineTransformer;
+import org.opensearch.dataprepper.core.peerforwarder.server.PeerForwarderServer;
+import org.opensearch.dataprepper.core.pipeline.Pipeline;
+import org.opensearch.dataprepper.core.pipeline.PipelineObserver;
+import org.opensearch.dataprepper.core.pipeline.PipelinesProvider;
+import org.opensearch.dataprepper.core.pipeline.server.DataPrepperServer;
 import org.opensearch.dataprepper.model.configuration.PipelinesDataFlowModel;
 import org.opensearch.dataprepper.model.plugin.PluginFactory;
-import org.opensearch.dataprepper.parser.PipelineTransformer;
-import org.opensearch.dataprepper.peerforwarder.server.PeerForwarderServer;
-import org.opensearch.dataprepper.pipeline.Pipeline;
-import org.opensearch.dataprepper.pipeline.PipelineObserver;
-import org.opensearch.dataprepper.pipeline.PipelinesProvider;
-import org.opensearch.dataprepper.pipeline.server.DataPrepperServer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.context.annotation.Lazy;
@@ -106,6 +108,8 @@ private void shutdownPipelines() {
 
     /**
      * Triggers the shutdown of all configured valid pipelines.
+     * @param shutdownOptions {@link DataPrepperShutdownOptions} to control the behavior of the shutdown process
+     *                        e.g. timeout, graceful shutdown, etc.
      */
     public void shutdownPipelines(final DataPrepperShutdownOptions shutdownOptions) {
         transformationPipelines.forEach((name, pipeline) -> {

diff --git a/...wledgements/AcknowledgementAppConfig.java → ...wledgements/AcknowledgementAppConfig.java b/...wledgements/AcknowledgementAppConfig.java → ...wledgements/AcknowledgementAppConfig.java
@@ -3,13 +3,13 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package org.opensearch.dataprepper.acknowledgements;
+package org.opensearch.dataprepper.core.acknowledgements;
 
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
-import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadFactory;
 
 @Configuration

diff --git a/...ledgements/AcknowledgementSetMonitor.java → ...ledgements/AcknowledgementSetMonitor.java b/...ledgements/AcknowledgementSetMonitor.java → ...ledgements/AcknowledgementSetMonitor.java
@@ -3,17 +3,16 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package org.opensearch.dataprepper.acknowledgements;
+package org.opensearch.dataprepper.core.acknowledgements;
 
 import org.opensearch.dataprepper.model.acknowledgements.AcknowledgementSet;
-
-import java.util.concurrent.locks.ReentrantLock;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.Set;
-import java.util.HashSet;
-
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReentrantLock;
 /**
  * AcknowledgementSetMonitor - monitors the acknowledgement sets for completion/expiration
  * <p>

diff --git a/...ents/AcknowledgementSetMonitorThread.java → ...ents/AcknowledgementSetMonitorThread.java b/...ents/AcknowledgementSetMonitorThread.java → ...ents/AcknowledgementSetMonitorThread.java
@@ -3,7 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package org.opensearch.dataprepper.acknowledgements;
+package org.opensearch.dataprepper.core.acknowledgements;
 
 import java.time.Duration;
 

diff --git a/...cknowledgements/CallbackTheadFactory.java → ...cknowledgements/CallbackTheadFactory.java b/...cknowledgements/CallbackTheadFactory.java → ...cknowledgements/CallbackTheadFactory.java
@@ -3,7 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package org.opensearch.dataprepper.acknowledgements;
+package org.opensearch.dataprepper.core.acknowledgements;
 
 import java.util.Objects;
 import java.util.concurrent.ThreadFactory;

diff --git a/...ledgements/DefaultAcknowledgementSet.java → ...ledgements/DefaultAcknowledgementSet.java b/...ledgements/DefaultAcknowledgementSet.java → ...ledgements/DefaultAcknowledgementSet.java
@@ -3,14 +3,14 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package org.opensearch.dataprepper.acknowledgements;
+package org.opensearch.dataprepper.core.acknowledgements;
 
 import org.opensearch.dataprepper.model.acknowledgements.AcknowledgementSet;
 import org.opensearch.dataprepper.model.acknowledgements.ProgressCheck;
+import org.opensearch.dataprepper.model.event.DefaultEventHandle;
 import org.opensearch.dataprepper.model.event.Event;
 import org.opensearch.dataprepper.model.event.EventHandle;
 import org.opensearch.dataprepper.model.event.InternalEventHandle;
-import org.opensearch.dataprepper.model.event.DefaultEventHandle;
 import org.opensearch.dataprepper.model.event.JacksonEvent;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -19,12 +19,12 @@
 import java.time.Instant;
 import java.util.HashMap;
 import java.util.Map;
-import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.locks.ReentrantLock;
-import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
 public class DefaultAcknowledgementSet implements AcknowledgementSet {

diff --git a/...nts/DefaultAcknowledgementSetManager.java → ...nts/DefaultAcknowledgementSetManager.java b/...nts/DefaultAcknowledgementSetManager.java → ...nts/DefaultAcknowledgementSetManager.java
@@ -3,11 +3,11 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package org.opensearch.dataprepper.acknowledgements;
+package org.opensearch.dataprepper.core.acknowledgements;
 
+import org.opensearch.dataprepper.metrics.PluginMetrics;
 import org.opensearch.dataprepper.model.acknowledgements.AcknowledgementSet;
 import org.opensearch.dataprepper.model.acknowledgements.AcknowledgementSetManager;
-import org.opensearch.dataprepper.metrics.PluginMetrics;
 
 import javax.inject.Inject;
 import javax.inject.Named;

diff --git a/...nts/DefaultAcknowledgementSetMetrics.java → ...nts/DefaultAcknowledgementSetMetrics.java b/...nts/DefaultAcknowledgementSetMetrics.java → ...nts/DefaultAcknowledgementSetMetrics.java
@@ -3,10 +3,10 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package org.opensearch.dataprepper.acknowledgements;
+package org.opensearch.dataprepper.core.acknowledgements;
 
-import org.opensearch.dataprepper.metrics.PluginMetrics;
 import io.micrometer.core.instrument.Counter;
+import org.opensearch.dataprepper.metrics.PluginMetrics;
 
 public class DefaultAcknowledgementSetMetrics {
     static final String CREATED_METRIC_NAME = "numberOfAcknowledgementSetsCreated";

diff --git a/...cknowledgements/DefaultProgressCheck.java → ...cknowledgements/DefaultProgressCheck.java b/...cknowledgements/DefaultProgressCheck.java → ...cknowledgements/DefaultProgressCheck.java
@@ -3,7 +3,7 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package org.opensearch.dataprepper.acknowledgements;
+package org.opensearch.dataprepper.core.acknowledgements;
 
 import org.opensearch.dataprepper.model.acknowledgements.ProgressCheck;
 

diff --git a/...ts/InactiveAcknowledgementSetManager.java → ...ts/InactiveAcknowledgementSetManager.java b/...ts/InactiveAcknowledgementSetManager.java → ...ts/InactiveAcknowledgementSetManager.java
@@ -3,12 +3,13 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package org.opensearch.dataprepper.acknowledgements;
+package org.opensearch.dataprepper.core.acknowledgements;
 
-import org.opensearch.dataprepper.model.acknowledgements.AcknowledgementSetManager;
 import org.opensearch.dataprepper.model.acknowledgements.AcknowledgementSet;
-import java.util.function.Consumer;
+import org.opensearch.dataprepper.model.acknowledgements.AcknowledgementSetManager;
+
 import java.time.Duration;
+import java.util.function.Consumer;
 
 public class InactiveAcknowledgementSetManager implements AcknowledgementSetManager {
     private static InactiveAcknowledgementSetManager theInstance;

diff --git a/...pper/breaker/CircuitBreakerAppConfig.java → ...core/breaker/CircuitBreakerAppConfig.java b/...pper/breaker/CircuitBreakerAppConfig.java → ...core/breaker/CircuitBreakerAppConfig.java
@@ -3,11 +3,11 @@
  * SPDX-License-Identifier: Apache-2.0
  */
 
-package org.opensearch.dataprepper.breaker;
+package org.opensearch.dataprepper.core.breaker;
 
+import org.opensearch.dataprepper.core.parser.model.CircuitBreakerConfig;
+import org.opensearch.dataprepper.core.parser.model.DataPrepperConfiguration;
 import org.opensearch.dataprepper.model.breaker.CircuitBreaker;
-import org.opensearch.dataprepper.parser.model.CircuitBreakerConfig;
-import org.opensearch.dataprepper.parser.model.DataPrepperConfiguration;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;