Re-enable ZeroBuffer

Signed-off-by: Mohammed Aghil Puthiyottil <57040494+MohammedAghil@users.noreply.github.com>

Author: MohammedAghil

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

@@ -12,6 +12,8 @@
 import org.opensearch.dataprepper.core.peerforwarder.PeerForwardingProcessorDecorator;
 import org.opensearch.dataprepper.core.pipeline.Pipeline;
 import org.opensearch.dataprepper.core.pipeline.PipelineConnector;
+import org.opensearch.dataprepper.core.pipeline.PipelineRunnerImpl;
+import org.opensearch.dataprepper.core.pipeline.SupportsPipelineRunner;
 import org.opensearch.dataprepper.core.pipeline.router.Router;
 import org.opensearch.dataprepper.core.pipeline.router.RouterFactory;
 import org.opensearch.dataprepper.core.sourcecoordination.SourceCoordinatorFactory;
@@ -228,10 +230,25 @@ private void buildPipelineFromConfiguration(
                     dataPrepperConfiguration.getProcessorShutdownTimeout(), dataPrepperConfiguration.getSinkShutdownTimeout(),
                     getPeerForwarderDrainTimeout(dataPrepperConfiguration));
 
-            // TODO: Re-enable zero-buffer
-            //if (pipelineDefinedBuffer instanceof SupportsPipelineRunner) {
-            //    ((SupportsPipelineRunner) pipelineDefinedBuffer).setPipelineRunner(new PipelineRunnerImpl(pipeline, processors));
-            //}
+            if (pipelineDefinedBuffer instanceof SupportsPipelineRunner) {
+                // Check if there are any processor sets with  @SingleThread processors
+                boolean hasSingleThreadedProcessors = processorSets.stream().anyMatch(processorSet -> processorSet.size() > 1);
+                // Only allow ZeroBuffer for single-threaded pipelines with no @SingleThread processors
+                if (processorThreads == 1 && !hasSingleThreadedProcessors) {
+                    ((SupportsPipelineRunner) pipelineDefinedBuffer).setPipelineRunner(
+                        new PipelineRunnerImpl(pipeline, pipeline.getProcessors()));
+                } else {
+                    if (hasSingleThreadedProcessors) {
+                        throw new IllegalStateException(
+                            "ZeroBuffer cannot be used with @SingleThread processors. " +
+                            "Pipeline [" + pipelineName + "] contains one or more @SingleThread processors.");
+                    } else {
+                        throw new IllegalStateException(
+                            "ZeroBuffer cannot be used with multiple processor threads. " +
+                            "Pipeline [" + pipelineName + "] is configured with " + processorThreads + " threads.");
+                    }
+                }
+            }
 
             pipelineMap.put(pipelineName, pipeline);
         } catch (Exception ex) {

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

@@ -71,6 +71,10 @@ public class TestDataProvider {
     public static final String VALID_PEER_FORWARDER_WITH_ACM_SSL_CONFIG_FILE = "src/test/resources/valid_peer_forwarder_config_with_acm_ssl.yml";
     public static final String VALID_DATA_PREPPER_CONFIG_WITH_METRIC_FILTER = "src/test/resources/valid_data_prepper_config_with_metric_filter.yml";
     public static final String INVALID_DATA_PREPPER_CONFIG_WITH_METRIC_FILTER = "src/test/resources/invalid_data_prepper_config_with_metric_filter.yml";
+    public static final String VALID_ZERO_BUFFER_SINGLE_THREAD_CONFIG_FILE = "src/test/resources/valid_zero_buffer_single_thread.yml";
+    public static final String INVALID_ZERO_BUFFER_MULTIPLE_THREADS_CONFIG_FILE = "src/test/resources/invalid_zero_buffer_multiple_threads.yml";
+    public static final String INVALID_ZERO_BUFFER_WITH_SINGLE_THREAD_PROCESSOR_CONFIG_FILE = "src/test/resources/invalid_zero_buffer_with_single_thread_processor.yml";
+    public static final String INVALID_ZERO_BUFFER_MULTIPLE_THREADS_NO_SINGLE_THREAD_PROCESSORS_CONFIG_FILE = "src/test/resources/invalid_zero_buffer_multiple_threads_no_single_thread_processors.yml";
     public static Set<String> VALID_MULTIPLE_PIPELINE_NAMES = new HashSet<>(Arrays.asList("test-pipeline-1",
             "test-pipeline-2", "test-pipeline-3"));
 }

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

@@ -67,6 +67,7 @@
 import static org.hamcrest.Matchers.hasKey;
 import static org.hamcrest.core.Is.is;
 import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
 import static org.mockito.ArgumentMatchers.anyString;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.mock;
@@ -603,4 +604,39 @@ private Map<String, Map<String, PeerForwarderReceiveBuffer<Record<Event>>>> gene
 
         return bufferMap;
     }
+
+    @Test
+    void parseConfiguration_with_zero_buffer_and_single_worker_thread_creates_pipeline_successfully() {
+        mockDataPrepperConfigurationAccesses();
+        final PipelineTransformer pipelineTransformer =
+                createObjectUnderTest(TestDataProvider.VALID_ZERO_BUFFER_SINGLE_THREAD_CONFIG_FILE);
+        final Map<String, Pipeline> pipelineMap = pipelineTransformer.transformConfiguration();
+
+        assertThat(pipelineMap.size(), equalTo(1));
+        assertThat(pipelineMap, hasKey("simple-pipeline"));
+        assertThat(pipelineMap.isEmpty(), equalTo(false));
+
+        verifyDataPrepperConfigurationAccesses();
+        verify(dataPrepperConfiguration).getPipelineExtensions();
+    }
+
+    @ParameterizedTest
+    @MethodSource("provideInvalidZeroBufferConfigFiles")
+    void parseConfiguration_with_invalid_zero_buffer_configurations_results_in_pipeline_creation_failure(String configFile) {
+        mockDataPrepperConfigurationAccesses();
+        final PipelineTransformer pipelineTransformer = createObjectUnderTest(configFile);
+        final Map<String, Pipeline> pipelineMap = pipelineTransformer.transformConfiguration();
+
+        assertThat(pipelineMap.isEmpty(), equalTo(true));
+
+        verify(dataPrepperConfiguration).getPipelineExtensions();
+    }
+
+    private static Stream<Arguments> provideInvalidZeroBufferConfigFiles() {
+        return Stream.of(
+                Arguments.of(TestDataProvider.INVALID_ZERO_BUFFER_MULTIPLE_THREADS_CONFIG_FILE),
+                Arguments.of(TestDataProvider.INVALID_ZERO_BUFFER_WITH_SINGLE_THREAD_PROCESSOR_CONFIG_FILE),
+                Arguments.of(TestDataProvider.INVALID_ZERO_BUFFER_MULTIPLE_THREADS_NO_SINGLE_THREAD_PROCESSORS_CONFIG_FILE)
+        );
+    }
 }

data-prepper-core/src/test/resources/invalid_zero_buffer_multiple_threads.yml

@@ -0,0 +1,8 @@
+simple-sample-pipeline:
+  workers: 2
+  source:
+    random:
+  buffer:
+    zero:
+  sink:
+    - stdout:

data-prepper-core/src/test/resources/invalid_zero_buffer_multiple_threads_no_single_thread_processors.yml

@@ -0,0 +1,11 @@
+simple-pipeline:
+  workers: 2
+  source:
+    random:
+  buffer:
+    zero:
+  processor:
+    - string_converter:
+        upper_case: true
+  sink:
+    - stdout:

data-prepper-core/src/test/resources/invalid_zero_buffer_with_single_thread_processor.yml

@@ -0,0 +1,10 @@
+simple-pipeline:
+  workers: 2
+  source:
+    random:
+  buffer:
+    zero:
+  processor:
+    - test_processor:
+  sink:
+    - stdout:

data-prepper-core/src/test/resources/valid_zero_buffer_single_thread.yml

@@ -0,0 +1,8 @@
+simple-pipeline:
+  workers: 1
+  source:
+    random:
+  buffer:
+    zero:
+  sink:
+    - stdout: