Adds delay processor for adding delays in the processor chain

data-prepper-plugins/common/src/main/java/org/opensearch/dataprepper/plugins/processor/DelayProcessor.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.opensearch.dataprepper.model.annotations.DataPrepperPlugin;
+import org.opensearch.dataprepper.model.annotations.DataPrepperPluginConstructor;
+import org.opensearch.dataprepper.model.processor.Processor;
+import org.opensearch.dataprepper.model.record.Record;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Duration;
+import java.util.Collection;
+
+@DataPrepperPlugin(name = "delay", pluginType = Processor.class, pluginConfigurationType = DelayProcessor.Configuration.class)
+public class DelayProcessor implements Processor<Record<?>, Record<?>> {
+    private static final Logger LOG = LoggerFactory.getLogger(DelayProcessor.class);
+    private final Duration delayDuration;
+
+    @DataPrepperPluginConstructor
+    public DelayProcessor(final Configuration configuration) {
+        delayDuration = configuration.getDelayFor();
+        LOG.info("Delay processor configured for {}. The pipeline will delay for this time. This is typically only for testing or debugging.", delayDuration);
+    }
+
+    @Override
+    public Collection<Record<?>> execute(final Collection<Record<?>> records) {
+        try {
+            Thread.sleep(delayDuration.toMillis());
+        } catch (final InterruptedException ex) {
+            LOG.error("Interrupted during delay processor", ex);
+        }
+        return records;
+    }
+
+    @Override
+    public void prepareForShutdown() {
+
+    }
+
+    @Override
+    public boolean isReadyForShutdown() {
+        return true;
+    }
+
+    @Override
+    public void shutdown() {
+
+    }
+
+    public static class Configuration {
+        @JsonProperty("for")
+        private Duration delayFor = Duration.ofSeconds(1);
+
+        public Duration getDelayFor() {
+            return delayFor;
+        }
+    }
+}

data-prepper-plugins/common/src/test/java/org/opensearch/dataprepper/plugins/processor/DelayProcessorTest.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.dataprepper.plugins.processor;
+
+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.model.record.Record;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.List;
+
+import static org.hamcrest.CoreMatchers.equalTo;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.greaterThanOrEqualTo;
+import static org.hamcrest.Matchers.lessThanOrEqualTo;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class DelayProcessorTest {
+    @Mock
+    private DelayProcessor.Configuration configuration;
+    private List<Record<?>> records;
+    private Duration delayDuration;
+
+    @BeforeEach
+    void setUp() {
+        delayDuration = Duration.ofMillis(50);
+        when(configuration.getDelayFor()).thenReturn(delayDuration);
+
+        records = List.of(mock(Record.class), mock(Record.class), mock(Record.class));
+    }
+
+    private DelayProcessor createObjectUnderTest() {
+        return new DelayProcessor(configuration);
+    }
+
+    @Test
+    void isReadyForShutdown_returns_true() {
+        assertThat(createObjectUnderTest().isReadyForShutdown(), equalTo(true));
+    }
+
+    @Test
+    void execute_returns_input_records() {
+        assertThat(createObjectUnderTest().execute(records),
+                equalTo(records));
+    }
+
+    @Test
+    void execute_takes_at_least_as_long_as_defined() {
+        final Instant before = Instant.now();
+        createObjectUnderTest().execute(records);
+        final Instant after = Instant.now();
+
+        assertThat(Duration.between(before, after), greaterThanOrEqualTo(delayDuration));
+        assertThat(Duration.between(before, after), lessThanOrEqualTo(delayDuration.multipliedBy(2)));
+    }
+}