Remove SingleThread annotation from Drop Events Processor (opensearch-project#6417)

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

Author: kkondaka

data-prepper-plugins/drop-events-processor/src/main/java/org/opensearch/dataprepper/plugins/processor/drop/DropEventsProcessor.java

@@ -12,7 +12,6 @@
 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.annotations.SingleThread;
 import org.opensearch.dataprepper.model.event.Event;
 import org.opensearch.dataprepper.model.plugin.InvalidPluginConfigurationException;
 import org.opensearch.dataprepper.model.processor.AbstractProcessor;
@@ -24,7 +23,6 @@
 import java.util.Collections;
 import java.util.stream.Collectors;
 
-@SingleThread
 @DataPrepperPlugin(name = "drop_events", pluginType = Processor.class, pluginConfigurationType = DropEventProcessorConfig.class)
 public class DropEventsProcessor extends AbstractProcessor<Record<Event>, Record<Event>> {
 

data-prepper-plugins/drop-events-processor/src/test/java/org/opensearch/dataprepper/plugins/processor/drop/DropEventsProcessorIT.java

@@ -19,24 +19,25 @@
 import org.opensearch.dataprepper.test.plugins.PluginConfigurationFile;
 import org.opensearch.dataprepper.test.plugins.junit.BaseDataPrepperPluginStandardTestSuite;
 
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import java.util.stream.Collectors;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.CoreMatchers.notNullValue;
 import static org.hamcrest.MatcherAssert.assertThat;
 
 @DataPrepperPluginTest(pluginName = "drop_events", pluginType = Processor.class)
 class DropEventsProcessorIT extends BaseDataPrepperPluginStandardTestSuite {
-    @Test
-    void drops_records_when_value_is_empty_string(
-            @PluginConfigurationFile("drop_when_value_is_empty_string.yaml") final Processor<Record<Event>, Record<Event>> objectUnderTest,
-            final EventFactory eventFactory) {
+    private static final int MAX_THREADS = 5;
 
+    private void createEventsAndExecute(final Processor<Record<Event>, Record<Event>> objectUnderTest,
+            final EventFactory eventFactory) {
         final List<Event> allEvents = new LinkedList<>();
         final List<Event> expectedEventsAfterProcessor = new LinkedList<>();
         for(int i = 0; i < 5; i++) {
@@ -69,4 +70,33 @@ void drops_records_when_value_is_empty_string(
 
         assertThat(outputEvents, equalTo(expectedEventsAfterProcessor));
     }
-}
+
+    @Test
+    void drops_records_when_value_is_empty_string(
+            @PluginConfigurationFile("drop_when_value_is_empty_string.yaml") final Processor<Record<Event>, Record<Event>> objectUnderTest,
+            final EventFactory eventFactory) {
+        createEventsAndExecute(objectUnderTest, eventFactory);
+    }
+
+    @Test
+    void drops_records_when_value_is_empty_string_with_multiple_threads(
+            @PluginConfigurationFile("drop_when_value_is_empty_string.yaml") final Processor<Record<Event>, Record<Event>> objectUnderTest,
+            final EventFactory eventFactory) throws Exception {
+
+        AtomicInteger numSuccess = new AtomicInteger(0);
+        List<Thread> threads = new ArrayList<>();
+        for (int i = 0; i < MAX_THREADS; i++) {
+            threads.add(new Thread(() -> {
+                createEventsAndExecute(objectUnderTest, eventFactory);
+                numSuccess.incrementAndGet();
+            }));
+        }
+        for (int i = 0; i < MAX_THREADS; i++) {
+            threads.get(i).start();
+        }
+        for (int i = 0; i < MAX_THREADS; i++) {
+            threads.get(i).join();
+        }
+        assertThat(numSuccess.get(), equalTo(MAX_THREADS));
+    }
+}