diff --git a/data-prepper-core/src/test/java/com/amazon/dataprepper/plugin/ClasspathPluginProviderTest.java b/data-prepper-core/src/test/java/com/amazon/dataprepper/plugin/ClasspathPluginProviderTest.java
index b89701437f..344403a888 100644
--- a/data-prepper-core/src/test/java/com/amazon/dataprepper/plugin/ClasspathPluginProviderTest.java
+++ b/data-prepper-core/src/test/java/com/amazon/dataprepper/plugin/ClasspathPluginProviderTest.java
@@ -4,7 +4,7 @@
 import com.amazon.dataprepper.model.sink.Sink;
 import com.amazon.dataprepper.model.source.Source;
 import com.amazon.dataprepper.plugins.TestSink;
-import com.amazon.dataprepper.plugins.TestSinkUpdated;
+import com.amazon.dataprepper.plugins.TestSinkDeprecatedApproach;
 import com.amazon.dataprepper.plugins.TestSource;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Nested;
@@ -85,7 +85,7 @@ class WithPredefinedPlugins {
         void setUp() {
             given(reflections.getTypesAnnotatedWith(DataPrepperPlugin.class))
                     .willReturn(new HashSet<>(Arrays.asList(
-                            TestSink.class, TestSource.class, TestSinkUpdated.class)));
+                            TestSink.class, TestSource.class, TestSinkDeprecatedApproach.class)));
         }
 
         @Test
@@ -104,18 +104,18 @@ void findPlugin_should_return_empty_if_plugin_found_for_another_type() {
 
         @Test
         void findPlugin_should_return_plugin_if_found_for_name_and_type() {
-            final Optional<Class<? extends Sink>> optionalPlugin = createObjectUnderTest().findPluginClass(Sink.class, "test_sink");
+            final Optional<Class<? extends Sink>> optionalPlugin = createObjectUnderTest().findPluginClass(Sink.class, "test_sink_deprecated_type");
             assertThat(optionalPlugin, notNullValue());
             assertThat(optionalPlugin.isPresent(), equalTo(true));
-            assertThat(optionalPlugin.get(), equalTo(TestSink.class));
+            assertThat(optionalPlugin.get(), equalTo(TestSinkDeprecatedApproach.class));
         }
 
         @Test
         void findPlugin_should_return_plugin_if_found_for_name_and_type_using_pluginType() {
-            final Optional<Class<? extends Sink>> optionalPlugin = createObjectUnderTest().findPluginClass(Sink.class, "test_sink_updated");
+            final Optional<Class<? extends Sink>> optionalPlugin = createObjectUnderTest().findPluginClass(Sink.class, "test_sink");
             assertThat(optionalPlugin, notNullValue());
             assertThat(optionalPlugin.isPresent(), equalTo(true));
-            assertThat(optionalPlugin.get(), equalTo(TestSinkUpdated.class));
+            assertThat(optionalPlugin.get(), equalTo(TestSink.class));
         }
     }
 }
\ No newline at end of file
diff --git a/data-prepper-core/src/test/java/com/amazon/dataprepper/plugins/TestSink.java b/data-prepper-core/src/test/java/com/amazon/dataprepper/plugins/TestSink.java
index 260b8e26bd..c8b6d1bbae 100644
--- a/data-prepper-core/src/test/java/com/amazon/dataprepper/plugins/TestSink.java
+++ b/data-prepper-core/src/test/java/com/amazon/dataprepper/plugins/TestSink.java
@@ -11,7 +11,6 @@
 
 package com.amazon.dataprepper.plugins;
 
-import com.amazon.dataprepper.model.PluginType;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.record.Record;
 import com.amazon.dataprepper.model.sink.Sink;
@@ -21,7 +20,7 @@
 import java.util.List;
 import java.util.stream.Collectors;
 
-@DataPrepperPlugin(name = "test_sink", type = PluginType.SINK)
+@DataPrepperPlugin(name = "test_sink", pluginType = Sink.class)
 public class TestSink implements Sink<Record<String>> {
     private final List<Record<String>> collectedRecords;
     private final boolean failSinkForTest;
diff --git a/data-prepper-core/src/test/java/com/amazon/dataprepper/plugins/TestSinkUpdated.java b/data-prepper-core/src/test/java/com/amazon/dataprepper/plugins/TestSinkDeprecatedApproach.java
similarity index 65%
rename from data-prepper-core/src/test/java/com/amazon/dataprepper/plugins/TestSinkUpdated.java
rename to data-prepper-core/src/test/java/com/amazon/dataprepper/plugins/TestSinkDeprecatedApproach.java
index c413a76382..6d00890492 100644
--- a/data-prepper-core/src/test/java/com/amazon/dataprepper/plugins/TestSinkUpdated.java
+++ b/data-prepper-core/src/test/java/com/amazon/dataprepper/plugins/TestSinkDeprecatedApproach.java
@@ -1,13 +1,14 @@
 package com.amazon.dataprepper.plugins;
 
+import com.amazon.dataprepper.model.PluginType;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.record.Record;
 import com.amazon.dataprepper.model.sink.Sink;
 
 import java.util.Collection;
 
-@DataPrepperPlugin(name = "test_sink_updated", pluginType = Sink.class)
-public class TestSinkUpdated implements Sink<Record<String>> {
+@DataPrepperPlugin(name = "test_sink_deprecated_type", type = PluginType.SINK)
+public class TestSinkDeprecatedApproach implements Sink<Record<String>> {
     @Override
     public void output(final Collection<Record<String>> records) {
 
diff --git a/data-prepper-core/src/test/java/com/amazon/dataprepper/plugins/TestSource.java b/data-prepper-core/src/test/java/com/amazon/dataprepper/plugins/TestSource.java
index 7334411677..e4e30d30c7 100644
--- a/data-prepper-core/src/test/java/com/amazon/dataprepper/plugins/TestSource.java
+++ b/data-prepper-core/src/test/java/com/amazon/dataprepper/plugins/TestSource.java
@@ -11,7 +11,6 @@
 
 package com.amazon.dataprepper.plugins;
 
-import com.amazon.dataprepper.model.PluginType;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.buffer.Buffer;
 import com.amazon.dataprepper.model.record.Record;
@@ -23,7 +22,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
-@DataPrepperPlugin(name = "test_source", type = PluginType.SOURCE)
+@DataPrepperPlugin(name = "test_source", pluginType = Source.class)
 public class TestSource implements Source<Record<String>> {
     public static final List<Record<String>> TEST_DATA = Stream.of("TEST")
             .map(Record::new).collect(Collectors.toList());
diff --git a/data-prepper-plugins/blocking-buffer/src/main/java/com/amazon/dataprepper/plugins/buffer/blockingbuffer/BlockingBuffer.java b/data-prepper-plugins/blocking-buffer/src/main/java/com/amazon/dataprepper/plugins/buffer/blockingbuffer/BlockingBuffer.java
index 2d5eb9bf24..f56bdbf448 100644
--- a/data-prepper-plugins/blocking-buffer/src/main/java/com/amazon/dataprepper/plugins/buffer/blockingbuffer/BlockingBuffer.java
+++ b/data-prepper-plugins/blocking-buffer/src/main/java/com/amazon/dataprepper/plugins/buffer/blockingbuffer/BlockingBuffer.java
@@ -11,14 +11,13 @@
 
 package com.amazon.dataprepper.plugins.buffer.blockingbuffer;
 
-import com.amazon.dataprepper.model.PluginType;
+import com.amazon.dataprepper.model.CheckpointState;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.buffer.AbstractBuffer;
 import com.amazon.dataprepper.model.buffer.Buffer;
 import com.amazon.dataprepper.model.buffer.SizeOverflowException;
 import com.amazon.dataprepper.model.configuration.PluginSetting;
 import com.amazon.dataprepper.model.record.Record;
-import com.amazon.dataprepper.model.CheckpointState;
 import com.google.common.base.Stopwatch;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -46,7 +45,7 @@
  * record is null. {@link #read(int)} retrieves and removes the batch of records from the head of the queue. The
  * batch size is defined/determined by the configuration attribute {@link #ATTRIBUTE_BATCH_SIZE} or the timeout parameter
  */
-@DataPrepperPlugin(name = "bounded_blocking", type = PluginType.BUFFER)
+@DataPrepperPlugin(name = "bounded_blocking", pluginType = Buffer.class)
 public class BlockingBuffer<T extends Record<?>> extends AbstractBuffer<T> {
     private static final Logger LOG = LoggerFactory.getLogger(BlockingBuffer.class);
     private static final int DEFAULT_BUFFER_CAPACITY = 512;
diff --git a/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/PluginRepository.java b/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/PluginRepository.java
index a1b54423c3..c0dd4f8071 100644
--- a/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/PluginRepository.java
+++ b/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/PluginRepository.java
@@ -21,6 +21,7 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Set;
 
 /**
@@ -51,7 +52,7 @@ private static void findPlugins() {
             final DataPrepperPlugin dataPrepperPluginAnnotation = annotatedClass
                     .getAnnotation(DataPrepperPlugin.class);
             final String pluginName = dataPrepperPluginAnnotation.name();
-            final PluginType pluginType = dataPrepperPluginAnnotation.type();
+            final PluginType pluginType = extractPluginType(dataPrepperPluginAnnotation);
             switch (pluginType) {
                 case SOURCE:
                     SOURCES.put(pluginName, (Class<Source>) annotatedClass);
@@ -69,6 +70,26 @@ private static void findPlugins() {
         }
     }
 
+    private static PluginType extractPluginType(final DataPrepperPlugin dataPrepperPluginAnnotation) {
+        PluginType pluginType = dataPrepperPluginAnnotation.type();
+        if(pluginType == PluginType.NONE) {
+            final Class<?> pluginClassType = dataPrepperPluginAnnotation.pluginType();
+            if(Objects.equals(pluginClassType, Source.class)) {
+                pluginType = PluginType.SOURCE;
+            }
+            else if(Objects.equals(pluginClassType, Buffer.class)) {
+                pluginType = PluginType.BUFFER;
+            }
+            else if(Objects.equals(pluginClassType, Prepper.class)) {
+                pluginType = PluginType.PREPPER;
+            }
+            else if(Objects.equals(pluginClassType, Sink.class)) {
+                pluginType = PluginType.SINK;
+            }
+        }
+        return pluginType;
+    }
+
     public static Class<Source> getSourceClass(final String name) {
         return SOURCES.get(name);
     }
diff --git a/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/prepper/NoOpPrepper.java b/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/prepper/NoOpPrepper.java
index 98f4fdd29b..5be7f0651f 100644
--- a/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/prepper/NoOpPrepper.java
+++ b/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/prepper/NoOpPrepper.java
@@ -11,7 +11,6 @@
 
 package com.amazon.dataprepper.plugins.prepper;
 
-import com.amazon.dataprepper.model.PluginType;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.configuration.PluginSetting;
 import com.amazon.dataprepper.model.prepper.Prepper;
@@ -19,7 +18,7 @@
 
 import java.util.Collection;
 
-@DataPrepperPlugin(name = "no-op", type = PluginType.PREPPER)
+@DataPrepperPlugin(name = "no-op", pluginType = Prepper.class)
 public class NoOpPrepper<InputT extends Record<?>> implements Prepper<InputT, InputT> {
 
     /**
diff --git a/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/prepper/StringPrepper.java b/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/prepper/StringPrepper.java
index 1ef2935308..18fd075319 100644
--- a/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/prepper/StringPrepper.java
+++ b/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/prepper/StringPrepper.java
@@ -11,7 +11,6 @@
 
 package com.amazon.dataprepper.plugins.prepper;
 
-import com.amazon.dataprepper.model.PluginType;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.configuration.PluginSetting;
 import com.amazon.dataprepper.model.prepper.Prepper;
@@ -24,7 +23,7 @@
  * An simple String implementation of {@link Prepper} which generates new Records with upper case or lowercase content. The current
  * simpler implementation does not handle errors (if any).
  */
-@DataPrepperPlugin(name = "string_converter", type = PluginType.PREPPER)
+@DataPrepperPlugin(name = "string_converter", pluginType = Prepper.class)
 public class StringPrepper implements Prepper<Record<String>, Record<String>> {
 
     public static final String UPPER_CASE = "upper_case";
diff --git a/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/sink/FileSink.java b/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/sink/FileSink.java
index 95728d5acc..3a9e458035 100644
--- a/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/sink/FileSink.java
+++ b/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/sink/FileSink.java
@@ -11,7 +11,6 @@
 
 package com.amazon.dataprepper.plugins.sink;
 
-import com.amazon.dataprepper.model.PluginType;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.configuration.PluginSetting;
 import com.amazon.dataprepper.model.record.Record;
@@ -26,7 +25,7 @@
 
 import static java.lang.String.format;
 
-@DataPrepperPlugin(name = "file", type = PluginType.SINK)
+@DataPrepperPlugin(name = "file", pluginType = Sink.class)
 public class FileSink implements Sink<Record<String>> {
     private static final String SAMPLE_FILE_PATH = "src/resources/file-test-sample-output.txt";
 
diff --git a/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/sink/StdOutSink.java b/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/sink/StdOutSink.java
index 6c6ba8a1f5..bbef2225c1 100644
--- a/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/sink/StdOutSink.java
+++ b/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/sink/StdOutSink.java
@@ -11,7 +11,6 @@
 
 package com.amazon.dataprepper.plugins.sink;
 
-import com.amazon.dataprepper.model.PluginType;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.configuration.PluginSetting;
 import com.amazon.dataprepper.model.record.Record;
@@ -20,7 +19,7 @@
 import java.util.Collection;
 import java.util.Iterator;
 
-@DataPrepperPlugin(name = "stdout", type = PluginType.SINK)
+@DataPrepperPlugin(name = "stdout", pluginType = Sink.class)
 public class StdOutSink implements Sink<Record<String>> {
 
     /**
diff --git a/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/source/FileSource.java b/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/source/FileSource.java
index a6024b90f8..a2b75aed09 100644
--- a/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/source/FileSource.java
+++ b/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/source/FileSource.java
@@ -11,7 +11,6 @@
 
 package com.amazon.dataprepper.plugins.source;
 
-import com.amazon.dataprepper.model.PluginType;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.buffer.Buffer;
 import com.amazon.dataprepper.model.configuration.PluginSetting;
@@ -30,7 +29,7 @@
 import static com.google.common.base.Preconditions.checkNotNull;
 import static java.lang.String.format;
 
-@DataPrepperPlugin(name = "file", type = PluginType.SOURCE)
+@DataPrepperPlugin(name = "file", pluginType = Source.class)
 public class FileSource implements Source<Record<String>> {
     private static final Logger LOG = LoggerFactory.getLogger(FileSource.class);
     private static final String ATTRIBUTE_PATH = "path";
diff --git a/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/source/RandomStringSource.java b/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/source/RandomStringSource.java
index 4f40a2458d..a46befa5ba 100644
--- a/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/source/RandomStringSource.java
+++ b/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/source/RandomStringSource.java
@@ -11,25 +11,25 @@
 
 package com.amazon.dataprepper.plugins.source;
 
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
-import com.amazon.dataprepper.model.PluginType;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.buffer.Buffer;
 import com.amazon.dataprepper.model.configuration.PluginSetting;
 import com.amazon.dataprepper.model.record.Record;
 import com.amazon.dataprepper.model.source.Source;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.util.UUID;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /**
  * Generates a random string every 500 milliseconds. Intended to be used for testing setups
  */
-@DataPrepperPlugin(name = "random", type = PluginType.SOURCE)
+@DataPrepperPlugin(name = "random", pluginType = Source.class)
 public class RandomStringSource implements Source<Record<String>> {
 
     private static Logger LOG = LoggerFactory.getLogger(RandomStringSource.class);
diff --git a/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/source/StdInSource.java b/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/source/StdInSource.java
index 0f714e07bb..264c9ac9fd 100644
--- a/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/source/StdInSource.java
+++ b/data-prepper-plugins/common/src/main/java/com/amazon/dataprepper/plugins/source/StdInSource.java
@@ -11,7 +11,6 @@
 
 package com.amazon.dataprepper.plugins.source;
 
-import com.amazon.dataprepper.model.PluginType;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.buffer.Buffer;
 import com.amazon.dataprepper.model.configuration.PluginSetting;
@@ -30,7 +29,7 @@
  * A simple source which reads data from console each line at a time. It exits when it reads case insensitive "exit"
  * from console or if Pipeline notifies to stop.
  */
-@DataPrepperPlugin(name = "stdin", type = PluginType.SOURCE)
+@DataPrepperPlugin(name = "stdin", pluginType = Source.class)
 public class StdInSource implements Source<Record<String>> {
     private static final Logger LOG = LoggerFactory.getLogger(StdInSource.class);
     private static final String ATTRIBUTE_TIMEOUT = "write_timeout";
diff --git a/data-prepper-plugins/common/src/test/java/com/amazon/dataprepper/plugins/ConstructorLessComponent.java b/data-prepper-plugins/common/src/test/java/com/amazon/dataprepper/plugins/ConstructorLessComponent.java
index ecfa1660ae..e0a189b6c4 100644
--- a/data-prepper-plugins/common/src/test/java/com/amazon/dataprepper/plugins/ConstructorLessComponent.java
+++ b/data-prepper-plugins/common/src/test/java/com/amazon/dataprepper/plugins/ConstructorLessComponent.java
@@ -11,7 +11,6 @@
 
 package com.amazon.dataprepper.plugins;
 
-import com.amazon.dataprepper.model.PluginType;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.buffer.Buffer;
 import com.amazon.dataprepper.model.record.Record;
@@ -19,7 +18,7 @@
 
 import java.util.concurrent.TimeoutException;
 
-@DataPrepperPlugin(name = "junit-test", type = PluginType.SOURCE)
+@DataPrepperPlugin(name = "junit-test", pluginType = Source.class)
 public class ConstructorLessComponent implements Source<Record<String>> {
 
     @Override
diff --git a/data-prepper-plugins/common/src/test/java/com/amazon/dataprepper/plugins/buffer/TestBuffer.java b/data-prepper-plugins/common/src/test/java/com/amazon/dataprepper/plugins/buffer/TestBuffer.java
index ab3fad32c0..5274e69a19 100644
--- a/data-prepper-plugins/common/src/test/java/com/amazon/dataprepper/plugins/buffer/TestBuffer.java
+++ b/data-prepper-plugins/common/src/test/java/com/amazon/dataprepper/plugins/buffer/TestBuffer.java
@@ -11,12 +11,11 @@
 
 package com.amazon.dataprepper.plugins.buffer;
 
-import com.amazon.dataprepper.model.PluginType;
+import com.amazon.dataprepper.model.CheckpointState;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.buffer.Buffer;
 import com.amazon.dataprepper.model.configuration.PluginSetting;
 import com.amazon.dataprepper.model.record.Record;
-import com.amazon.dataprepper.model.CheckpointState;
 
 import java.util.AbstractMap;
 import java.util.ArrayList;
@@ -27,7 +26,7 @@
 import java.util.Queue;
 import java.util.concurrent.TimeoutException;
 
-@DataPrepperPlugin(name = "test_buffer", type = PluginType.BUFFER)
+@DataPrepperPlugin(name = "test_buffer", pluginType = Buffer.class)
 public class TestBuffer implements Buffer<Record<String>> {
     private static final String ATTRIBUTE_BATCH_SIZE = "batch_size";
     private static final String ATTRIBUTE_IMITATE_TIMEOUT = "imitate_timeout";
diff --git a/data-prepper-plugins/common/src/test/java/com/amazon/dataprepper/plugins/prepper/TestPrepper.java b/data-prepper-plugins/common/src/test/java/com/amazon/dataprepper/plugins/prepper/TestPrepper.java
index c3e320a50f..6426f3e5cb 100644
--- a/data-prepper-plugins/common/src/test/java/com/amazon/dataprepper/plugins/prepper/TestPrepper.java
+++ b/data-prepper-plugins/common/src/test/java/com/amazon/dataprepper/plugins/prepper/TestPrepper.java
@@ -11,16 +11,16 @@
 
 package com.amazon.dataprepper.plugins.prepper;
 
-import com.amazon.dataprepper.model.PluginType;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.annotations.SingleThread;
 import com.amazon.dataprepper.model.configuration.PluginSetting;
 import com.amazon.dataprepper.model.prepper.Prepper;
 import com.amazon.dataprepper.model.record.Record;
+
 import java.util.Collection;
 
 @SingleThread
-@DataPrepperPlugin(name = "test_prepper", type = PluginType.PREPPER)
+@DataPrepperPlugin(name = "test_prepper", pluginType = Prepper.class)
 public class TestPrepper implements Prepper<Record<String>, Record<String>> {
     public boolean isShutdown = false;
 
diff --git a/data-prepper-plugins/common/src/test/java/com/amazon/dataprepper/plugins/sink/TestSink.java b/data-prepper-plugins/common/src/test/java/com/amazon/dataprepper/plugins/sink/TestSink.java
index 8eee27dc59..2cd93508d5 100644
--- a/data-prepper-plugins/common/src/test/java/com/amazon/dataprepper/plugins/sink/TestSink.java
+++ b/data-prepper-plugins/common/src/test/java/com/amazon/dataprepper/plugins/sink/TestSink.java
@@ -12,7 +12,6 @@
 package com.amazon.dataprepper.plugins.sink;
 
 
-import com.amazon.dataprepper.model.PluginType;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.record.Record;
 import com.amazon.dataprepper.model.sink.Sink;
@@ -22,7 +21,7 @@
 import java.util.List;
 import java.util.stream.Collectors;
 
-@DataPrepperPlugin(name = "test-sink", type = PluginType.SINK)
+@DataPrepperPlugin(name = "test-sink", pluginType = Sink.class)
 public class TestSink implements Sink<Record<String>> {
     private final List<Record<String>> collectedRecords;
 
diff --git a/data-prepper-plugins/common/src/test/java/com/amazon/dataprepper/plugins/source/TestSource.java b/data-prepper-plugins/common/src/test/java/com/amazon/dataprepper/plugins/source/TestSource.java
index 1c679b500b..4763b6e885 100644
--- a/data-prepper-plugins/common/src/test/java/com/amazon/dataprepper/plugins/source/TestSource.java
+++ b/data-prepper-plugins/common/src/test/java/com/amazon/dataprepper/plugins/source/TestSource.java
@@ -11,7 +11,6 @@
 
 package com.amazon.dataprepper.plugins.source;
 
-import com.amazon.dataprepper.model.PluginType;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.buffer.Buffer;
 import com.amazon.dataprepper.model.record.Record;
@@ -23,7 +22,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
-@DataPrepperPlugin(name = "test-source", type = PluginType.SOURCE)
+@DataPrepperPlugin(name = "test-source", pluginType = Source.class)
 public class TestSource implements Source<Record<String>> {
     public static final List<Record<String>> TEST_DATA = Stream.of("THIS", "IS", "TEST", "DATA")
             .map(Record::new).collect(Collectors.toList());
diff --git a/data-prepper-plugins/grok-prepper/src/main/java/com/amazon/dataprepper/plugins/prepper/grok/GrokPrepper.java b/data-prepper-plugins/grok-prepper/src/main/java/com/amazon/dataprepper/plugins/prepper/grok/GrokPrepper.java
index d11bc3f7b8..c4140cbc10 100644
--- a/data-prepper-plugins/grok-prepper/src/main/java/com/amazon/dataprepper/plugins/prepper/grok/GrokPrepper.java
+++ b/data-prepper-plugins/grok-prepper/src/main/java/com/amazon/dataprepper/plugins/prepper/grok/GrokPrepper.java
@@ -12,17 +12,14 @@
 package com.amazon.dataprepper.plugins.prepper.grok;
 
 
-import com.amazon.dataprepper.model.PluginType;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
-
 import com.amazon.dataprepper.model.configuration.PluginSetting;
-import com.amazon.dataprepper.model.record.Record;
 import com.amazon.dataprepper.model.prepper.AbstractPrepper;
-
+import com.amazon.dataprepper.model.prepper.Prepper;
+import com.amazon.dataprepper.model.record.Record;
 import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.core.type.TypeReference;
-
+import com.fasterxml.jackson.databind.ObjectMapper;
 import io.krakens.grok.api.Grok;
 import io.krakens.grok.api.GrokCompiler;
 import io.krakens.grok.api.Match;
@@ -53,7 +50,7 @@
 import java.util.stream.Collectors;
 
 
-@DataPrepperPlugin(name = "grok", type = PluginType.PREPPER)
+@DataPrepperPlugin(name = "grok", pluginType = Prepper.class)
 public class GrokPrepper extends AbstractPrepper<Record<String>, Record<String>> {
 
     private static final Logger LOG = LoggerFactory.getLogger(GrokPrepper.class);
diff --git a/data-prepper-plugins/http-source/src/main/java/com/amazon/dataprepper/plugins/source/loghttp/HTTPSource.java b/data-prepper-plugins/http-source/src/main/java/com/amazon/dataprepper/plugins/source/loghttp/HTTPSource.java
index fd68a5640e..03cfeef130 100644
--- a/data-prepper-plugins/http-source/src/main/java/com/amazon/dataprepper/plugins/source/loghttp/HTTPSource.java
+++ b/data-prepper-plugins/http-source/src/main/java/com/amazon/dataprepper/plugins/source/loghttp/HTTPSource.java
@@ -12,7 +12,6 @@
 package com.amazon.dataprepper.plugins.source.loghttp;
 
 import com.amazon.dataprepper.metrics.PluginMetrics;
-import com.amazon.dataprepper.model.PluginType;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.buffer.Buffer;
 import com.amazon.dataprepper.model.configuration.PluginSetting;
@@ -33,7 +32,7 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 
-@DataPrepperPlugin(name = "http", type = PluginType.SOURCE)
+@DataPrepperPlugin(name = "http", pluginType = Source.class)
 public class HTTPSource implements Source<Record<String>> {
     private static final Logger LOG = LoggerFactory.getLogger(HTTPSource.class);
 
diff --git a/data-prepper-plugins/opensearch/src/main/java/com/amazon/dataprepper/plugins/sink/opensearch/OpenSearchSink.java b/data-prepper-plugins/opensearch/src/main/java/com/amazon/dataprepper/plugins/sink/opensearch/OpenSearchSink.java
index 1d220e0599..68f69e1dd2 100644
--- a/data-prepper-plugins/opensearch/src/main/java/com/amazon/dataprepper/plugins/sink/opensearch/OpenSearchSink.java
+++ b/data-prepper-plugins/opensearch/src/main/java/com/amazon/dataprepper/plugins/sink/opensearch/OpenSearchSink.java
@@ -11,11 +11,11 @@
 
 package com.amazon.dataprepper.plugins.sink.opensearch;
 
-import com.amazon.dataprepper.model.PluginType;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.configuration.PluginSetting;
 import com.amazon.dataprepper.model.record.Record;
 import com.amazon.dataprepper.model.sink.AbstractSink;
+import com.amazon.dataprepper.model.sink.Sink;
 import com.amazon.dataprepper.plugins.sink.opensearch.index.IndexManager;
 import com.amazon.dataprepper.plugins.sink.opensearch.index.IndexManagerFactory;
 import com.amazon.dataprepper.plugins.sink.opensearch.index.IndexType;
@@ -27,7 +27,11 @@
 import org.opensearch.client.RequestOptions;
 import org.opensearch.client.RestHighLevelClient;
 import org.opensearch.common.unit.ByteSizeUnit;
-import org.opensearch.common.xcontent.*;
+import org.opensearch.common.xcontent.LoggingDeprecationHandler;
+import org.opensearch.common.xcontent.NamedXContentRegistry;
+import org.opensearch.common.xcontent.XContentFactory;
+import org.opensearch.common.xcontent.XContentParser;
+import org.opensearch.common.xcontent.XContentType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -41,7 +45,7 @@
 import java.util.Optional;
 import java.util.function.Supplier;
 
-@DataPrepperPlugin(name = "opensearch", type = PluginType.SINK)
+@DataPrepperPlugin(name = "opensearch", pluginType = Sink.class)
 public class OpenSearchSink extends AbstractSink<Record<String>> {
   public static final String BULKREQUEST_LATENCY = "bulkRequestLatency";
   public static final String BULKREQUEST_ERRORS = "bulkRequestErrors";
diff --git a/data-prepper-plugins/otel-trace-group-prepper/src/main/java/com/amazon/dataprepper/plugins/prepper/oteltracegroup/OTelTraceGroupPrepper.java b/data-prepper-plugins/otel-trace-group-prepper/src/main/java/com/amazon/dataprepper/plugins/prepper/oteltracegroup/OTelTraceGroupPrepper.java
index ea4faae810..fd8d747920 100644
--- a/data-prepper-plugins/otel-trace-group-prepper/src/main/java/com/amazon/dataprepper/plugins/prepper/oteltracegroup/OTelTraceGroupPrepper.java
+++ b/data-prepper-plugins/otel-trace-group-prepper/src/main/java/com/amazon/dataprepper/plugins/prepper/oteltracegroup/OTelTraceGroupPrepper.java
@@ -11,10 +11,10 @@
 
 package com.amazon.dataprepper.plugins.prepper.oteltracegroup;
 
-import com.amazon.dataprepper.model.PluginType;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.configuration.PluginSetting;
 import com.amazon.dataprepper.model.prepper.AbstractPrepper;
+import com.amazon.dataprepper.model.prepper.Prepper;
 import com.amazon.dataprepper.model.record.Record;
 import com.amazon.dataprepper.plugins.prepper.oteltracegroup.model.TraceGroup;
 import com.fasterxml.jackson.core.JsonProcessingException;
@@ -48,7 +48,7 @@
 import java.util.Set;
 import java.util.stream.Stream;
 
-@DataPrepperPlugin(name = "otel_trace_group_prepper", type = PluginType.PREPPER)
+@DataPrepperPlugin(name = "otel_trace_group_prepper", pluginType = Prepper.class)
 public class OTelTraceGroupPrepper extends AbstractPrepper<Record<String>, Record<String>> {
 
     public static final String RECORDS_IN_MISSING_TRACE_GROUP = "recordsInMissingTraceGroup";
diff --git a/data-prepper-plugins/otel-trace-raw-prepper/src/main/java/com/amazon/dataprepper/plugins/prepper/oteltrace/OTelTraceRawPrepper.java b/data-prepper-plugins/otel-trace-raw-prepper/src/main/java/com/amazon/dataprepper/plugins/prepper/oteltrace/OTelTraceRawPrepper.java
index cf3dc0ccf5..9b476d053a 100644
--- a/data-prepper-plugins/otel-trace-raw-prepper/src/main/java/com/amazon/dataprepper/plugins/prepper/oteltrace/OTelTraceRawPrepper.java
+++ b/data-prepper-plugins/otel-trace-raw-prepper/src/main/java/com/amazon/dataprepper/plugins/prepper/oteltrace/OTelTraceRawPrepper.java
@@ -11,10 +11,10 @@
 
 package com.amazon.dataprepper.plugins.prepper.oteltrace;
 
-import com.amazon.dataprepper.model.PluginType;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.configuration.PluginSetting;
 import com.amazon.dataprepper.model.prepper.AbstractPrepper;
+import com.amazon.dataprepper.model.prepper.Prepper;
 import com.amazon.dataprepper.model.record.Record;
 import com.amazon.dataprepper.plugins.prepper.oteltrace.model.OTelProtoHelper;
 import com.amazon.dataprepper.plugins.prepper.oteltrace.model.RawSpan;
@@ -45,7 +45,7 @@
 import java.util.concurrent.locks.ReentrantLock;
 
 
-@DataPrepperPlugin(name = "otel_trace_raw_prepper", type = PluginType.PREPPER)
+@DataPrepperPlugin(name = "otel_trace_raw_prepper", pluginType = Prepper.class)
 public class OTelTraceRawPrepper extends AbstractPrepper<Record<ExportTraceServiceRequest>, Record<String>> {
     private static final long SEC_TO_MILLIS = 1_000L;
     private static final Logger LOG = LoggerFactory.getLogger(OTelTraceRawPrepper.class);
diff --git a/data-prepper-plugins/otel-trace-source/src/main/java/com/amazon/dataprepper/plugins/source/oteltrace/OTelTraceSource.java b/data-prepper-plugins/otel-trace-source/src/main/java/com/amazon/dataprepper/plugins/source/oteltrace/OTelTraceSource.java
index a834a8974e..31d152cdff 100644
--- a/data-prepper-plugins/otel-trace-source/src/main/java/com/amazon/dataprepper/plugins/source/oteltrace/OTelTraceSource.java
+++ b/data-prepper-plugins/otel-trace-source/src/main/java/com/amazon/dataprepper/plugins/source/oteltrace/OTelTraceSource.java
@@ -12,16 +12,15 @@
 package com.amazon.dataprepper.plugins.source.oteltrace;
 
 import com.amazon.dataprepper.metrics.PluginMetrics;
-import com.amazon.dataprepper.model.PluginType;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.buffer.Buffer;
 import com.amazon.dataprepper.model.configuration.PluginSetting;
 import com.amazon.dataprepper.model.record.Record;
 import com.amazon.dataprepper.model.source.Source;
 import com.amazon.dataprepper.plugins.certificate.CertificateProvider;
-import com.amazon.dataprepper.plugins.source.oteltrace.certificate.CertificateProviderFactory;
 import com.amazon.dataprepper.plugins.certificate.model.Certificate;
 import com.amazon.dataprepper.plugins.health.HealthGrpcService;
+import com.amazon.dataprepper.plugins.source.oteltrace.certificate.CertificateProviderFactory;
 import com.linecorp.armeria.server.Server;
 import com.linecorp.armeria.server.ServerBuilder;
 import com.linecorp.armeria.server.grpc.GrpcService;
@@ -36,7 +35,7 @@
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executors;
 
-@DataPrepperPlugin(name = "otel_trace_source", type = PluginType.SOURCE)
+@DataPrepperPlugin(name = "otel_trace_source", pluginType = Source.class)
 public class OTelTraceSource implements Source<Record<ExportTraceServiceRequest>> {
     private static final Logger LOG = LoggerFactory.getLogger(OTelTraceSource.class);
     private final OTelTraceSourceConfig oTelTraceSourceConfig;
diff --git a/data-prepper-plugins/peer-forwarder/src/main/java/com/amazon/dataprepper/plugins/prepper/peerforwarder/PeerForwarder.java b/data-prepper-plugins/peer-forwarder/src/main/java/com/amazon/dataprepper/plugins/prepper/peerforwarder/PeerForwarder.java
index 659b2fb6cf..d2c6a21ea8 100644
--- a/data-prepper-plugins/peer-forwarder/src/main/java/com/amazon/dataprepper/plugins/prepper/peerforwarder/PeerForwarder.java
+++ b/data-prepper-plugins/peer-forwarder/src/main/java/com/amazon/dataprepper/plugins/prepper/peerforwarder/PeerForwarder.java
@@ -11,10 +11,10 @@
 
 package com.amazon.dataprepper.plugins.prepper.peerforwarder;
 
-import com.amazon.dataprepper.model.PluginType;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.configuration.PluginSetting;
 import com.amazon.dataprepper.model.prepper.AbstractPrepper;
+import com.amazon.dataprepper.model.prepper.Prepper;
 import com.amazon.dataprepper.model.record.Record;
 import com.amazon.dataprepper.plugins.prepper.peerforwarder.discovery.StaticPeerListProvider;
 import io.micrometer.core.instrument.Counter;
@@ -40,7 +40,7 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 
-@DataPrepperPlugin(name = "peer_forwarder", type = PluginType.PREPPER)
+@DataPrepperPlugin(name = "peer_forwarder", pluginType = Prepper.class)
 public class PeerForwarder extends AbstractPrepper<Record<ExportTraceServiceRequest>, Record<ExportTraceServiceRequest>> {
     public static final String REQUESTS = "requests";
     public static final String LATENCY = "latency";
diff --git a/data-prepper-plugins/service-map-stateful/src/main/java/com/amazon/dataprepper/plugins/prepper/ServiceMapStatefulPrepper.java b/data-prepper-plugins/service-map-stateful/src/main/java/com/amazon/dataprepper/plugins/prepper/ServiceMapStatefulPrepper.java
index 6830d063a9..406674f0c8 100644
--- a/data-prepper-plugins/service-map-stateful/src/main/java/com/amazon/dataprepper/plugins/prepper/ServiceMapStatefulPrepper.java
+++ b/data-prepper-plugins/service-map-stateful/src/main/java/com/amazon/dataprepper/plugins/prepper/ServiceMapStatefulPrepper.java
@@ -11,11 +11,11 @@
 
 package com.amazon.dataprepper.plugins.prepper;
 
-import com.amazon.dataprepper.model.PluginType;
 import com.amazon.dataprepper.model.annotations.DataPrepperPlugin;
 import com.amazon.dataprepper.model.annotations.SingleThread;
 import com.amazon.dataprepper.model.configuration.PluginSetting;
 import com.amazon.dataprepper.model.prepper.AbstractPrepper;
+import com.amazon.dataprepper.model.prepper.Prepper;
 import com.amazon.dataprepper.model.record.Record;
 import com.amazon.dataprepper.plugins.prepper.state.MapDbPrepperState;
 import com.fasterxml.jackson.databind.ObjectMapper;
@@ -39,7 +39,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 @SingleThread
-@DataPrepperPlugin(name = "service_map_stateful", type = PluginType.PREPPER)
+@DataPrepperPlugin(name = "service_map_stateful", pluginType = Prepper.class)
 public class ServiceMapStatefulPrepper extends AbstractPrepper<Record<ExportTraceServiceRequest>, Record<String>> {
 
     public static final String SPANS_DB_SIZE = "spansDbSize";