Addressed comments. Added more tests for 100% code coverage

Signed-off-by: Kondaka <krishkdk@amazon.com>

Author: kkondaka

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/buffer/AbstractBuffer.java

@@ -13,7 +13,7 @@
 import org.opensearch.dataprepper.model.configuration.PluginSetting;
 import org.opensearch.dataprepper.model.record.Record;
 import org.opensearch.dataprepper.model.event.Event;
-import org.opensearch.dataprepper.model.failures.FailurePipeline;
+import org.opensearch.dataprepper.model.pipeline.HeadlessPipeline;
 
 import java.time.Instant;
 import java.time.Duration;
@@ -39,7 +39,7 @@ public abstract class AbstractBuffer<T extends Record<?>> implements Buffer<T> {
     private final Timer latencyTimer;
     private final Timer readTimer;
     private final Timer checkpointTimer;
-    private FailurePipeline failurePipeline;
+    private HeadlessPipeline failurePipeline;
 
     public AbstractBuffer(final PluginSetting pluginSetting) {
         this(PluginMetrics.fromPluginSetting(pluginSetting), pluginSetting.getPipelineName());
@@ -65,11 +65,11 @@ private AbstractBuffer(final PluginMetrics pluginMetrics, final String pipelineN
     }
 
     @Override
-    public void setFailurePipeline(final FailurePipeline failurePipeline) {
+    public void setFailurePipeline(final HeadlessPipeline failurePipeline) {
         this.failurePipeline = failurePipeline;
     }
 
-    public FailurePipeline getFailurePipeline() {
+    public HeadlessPipeline getFailurePipeline() {
         return failurePipeline;
     }
 

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/buffer/Buffer.java

@@ -8,7 +8,7 @@
 import org.opensearch.dataprepper.model.CheckpointState;
 import org.opensearch.dataprepper.model.plugin.PluginComponentType;
 import org.opensearch.dataprepper.model.record.Record;
-import org.opensearch.dataprepper.model.failures.FailurePipeline;
+import org.opensearch.dataprepper.model.pipeline.HeadlessPipeline;
 
 import java.time.Duration;
 import java.util.Collection;
@@ -146,7 +146,7 @@ default void shutdown() {
      * @param failurePipeline failure pipeline
      * @since 2.12
      */
-    default void setFailurePipeline(final FailurePipeline failurePipeline) {
+    default void setFailurePipeline(final HeadlessPipeline failurePipeline) {
     }
 
 }

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/configuration/PipelineModel.java

@@ -77,7 +77,6 @@ public PipelineModel(
             @JsonProperty("sink") final List<SinkModel> sinks,
             @JsonProperty("workers") final Integer workers,
             @JsonProperty("delay") final Integer delay) {
-        checkArgument(Objects.nonNull(source), "Source must not be null");
         checkArgument(Objects.nonNull(sinks), "Sinks must not be null");
         checkArgument(sinks.size() > 0, "PipelineModel must include at least 1 sink");
 

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/failures/FailurePipeline.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.model.pipeline;
+
+import org.opensearch.dataprepper.model.event.Event;
+import org.opensearch.dataprepper.model.record.Record;
+
+import java.util.Collection;
+
+public interface HeadlessPipeline {
+    /**
+     * sets flag to indicate if acknowledgements are enabled
+     *
+     * @param acknowledgementsEnabled flag indicating acknowledgements are enabled
+     * @since 2.13
+     */
+    void setAcknowledgementsEnabled(final boolean acknowledgementsEnabled);
+
+    /**
+     * sends events to the headless pipeline
+     *
+     * @param records records to be sent to headless pipeline
+     * @since 2.13
+     */
+    void sendEvents(Collection<Record<Event>> events);
+    
+}

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/pipeline/HeadlessPipeline.java

@@ -11,7 +11,7 @@
 import org.opensearch.dataprepper.metrics.PluginMetrics;
 import org.opensearch.dataprepper.model.configuration.PluginSetting;
 import org.opensearch.dataprepper.model.record.Record;
-import org.opensearch.dataprepper.model.failures.FailurePipeline;
+import org.opensearch.dataprepper.model.pipeline.HeadlessPipeline;
 
 import java.util.Collection;
 
@@ -27,7 +27,7 @@ public abstract class AbstractProcessor<InputRecord extends Record<?>, OutputRec
     private final Counter recordsInCounter;
     private final Counter recordsOutCounter;
     private final Timer timeElapsedTimer;
-    private FailurePipeline failurePipeline;
+    private HeadlessPipeline failurePipeline;
 
     public AbstractProcessor(final PluginSetting pluginSetting) {
         pluginMetrics = PluginMetrics.fromPluginSetting(pluginSetting);
@@ -67,11 +67,11 @@ public Collection<OutputRecord> execute(final Collection<InputRecord> records) {
     public abstract Collection<OutputRecord> doExecute(Collection<InputRecord> records);
 
     @Override
-    public void setFailurePipeline(FailurePipeline failurePipeline) {
+    public void setFailurePipeline(HeadlessPipeline failurePipeline) {
         this.failurePipeline = failurePipeline;
     }
 
-    public FailurePipeline getFailurePipeline() {
+    public HeadlessPipeline getFailurePipeline() {
         return failurePipeline;
     }
 }

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/processor/AbstractProcessor.java

@@ -6,7 +6,7 @@
 package org.opensearch.dataprepper.model.processor;
 
 import org.opensearch.dataprepper.model.plugin.PluginComponentType;
-import org.opensearch.dataprepper.model.failures.FailurePipeline;
+import org.opensearch.dataprepper.model.pipeline.HeadlessPipeline;
 import org.opensearch.dataprepper.model.record.Record;
 
 import java.util.Collection;
@@ -66,7 +66,7 @@ default boolean holdsEvents() {
      * @param failurePipeline failure pipeline
      * @since 2.12
      */
-    default void setFailurePipeline(final FailurePipeline failurePipeline) {
+    default void setFailurePipeline(final HeadlessPipeline failurePipeline) {
     }
 
 }

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/processor/Processor.java

@@ -10,7 +10,7 @@
 import org.opensearch.dataprepper.metrics.MetricNames;
 import org.opensearch.dataprepper.metrics.PluginMetrics;
 import org.opensearch.dataprepper.model.configuration.PluginSetting;
-import org.opensearch.dataprepper.model.failures.FailurePipeline;
+import org.opensearch.dataprepper.model.pipeline.HeadlessPipeline;
 import org.opensearch.dataprepper.model.record.Record;
 import org.opensearch.dataprepper.model.event.Event;
 
@@ -30,7 +30,7 @@ public abstract class AbstractSink<T extends Record<?>> implements Sink<T> {
     private int maxRetries;
     private int waitTimeMs;
     private SinkThread sinkThread;
-    private FailurePipeline failurePipeline;
+    private HeadlessPipeline failurePipeline;
 
     public AbstractSink(final PluginSetting pluginSetting, int numRetries, int waitTimeMs) {
         this.pluginMetrics = PluginMetrics.fromPluginSetting(pluginSetting);
@@ -72,11 +72,11 @@ public void output(Collection<T> records) {
     }
 
     @Override
-    public void setFailurePipeline(final FailurePipeline failurePipeline) {
+    public void setFailurePipeline(final HeadlessPipeline failurePipeline) {
         this.failurePipeline = failurePipeline;
     }
 
-    public FailurePipeline getFailurePipeline() {
+    public HeadlessPipeline getFailurePipeline() {
         return failurePipeline;
     }
 

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/sink/AbstractSink.java

@@ -6,7 +6,7 @@
 package org.opensearch.dataprepper.model.sink;
 
 import org.opensearch.dataprepper.model.plugin.PluginComponentType;
-import org.opensearch.dataprepper.model.failures.FailurePipeline;
+import org.opensearch.dataprepper.model.pipeline.HeadlessPipeline;
 import org.opensearch.dataprepper.model.record.Record;
 
 import java.util.Collection;
@@ -55,7 +55,7 @@ default void updateLatencyMetrics(final Collection<T> events) {
      * @param failurePipeline failure pipeline
      * @since 2.12
      */
-    default void setFailurePipeline(final FailurePipeline failurePipeline) {
+    default void setFailurePipeline(final HeadlessPipeline failurePipeline) {
     }
 
 }

data-prepper-api/src/main/java/org/opensearch/dataprepper/model/sink/Sink.java

@@ -8,7 +8,7 @@
 import org.opensearch.dataprepper.model.buffer.Buffer;
 import org.opensearch.dataprepper.model.plugin.PluginComponentType;
 import org.opensearch.dataprepper.model.record.Record;
-import org.opensearch.dataprepper.model.failures.FailurePipeline;
+import org.opensearch.dataprepper.model.pipeline.HeadlessPipeline;
 import org.opensearch.dataprepper.model.codec.HasByteDecoder;
 
 /**
@@ -46,7 +46,7 @@ default boolean areAcknowledgementsEnabled() {
      * @param failurePipeline failure pipeline
      * @since 2.12
      */
-    default void setFailurePipeline(final FailurePipeline failurePipeline) {
+    default void setFailurePipeline(final HeadlessPipeline failurePipeline) {
     }
 
 }