Pass the pipeline identifier as part of the data prepper config.

data-prepper-plugins/kinesis-source/src/main/java/org/opensearch/dataprepper/plugins/kinesis/extension/KinesisLeaseConfig.java

@@ -17,4 +17,7 @@
 public class KinesisLeaseConfig {
     @JsonProperty("lease_coordination")
     private KinesisLeaseCoordinationTableConfig leaseCoordinationTable;
+
+    @JsonProperty("pipeline_identifier")
+    private String pipelineIdentifier;
 }

data-prepper-plugins/kinesis-source/src/main/java/org/opensearch/dataprepper/plugins/kinesis/source/KinesisService.java

@@ -60,7 +60,6 @@ public class KinesisService {
     private final String applicationName;
     private final String tableName;
     private final String kclMetricsNamespaceName;
-    private final String pipelineName;
     private final AcknowledgementSetManager acknowledgementSetManager;
     private final KinesisSourceConfig kinesisSourceConfig;
     private final KinesisAsyncClient kinesisClient;
@@ -96,8 +95,11 @@ public KinesisService(final KinesisSourceConfig kinesisSourceConfig,
         this.dynamoDbClient = kinesisClientFactory.buildDynamoDBClient(kinesisLeaseConfig.getLeaseCoordinationTable().getAwsRegion());
         this.kinesisClient = kinesisClientFactory.buildKinesisAsyncClient(kinesisSourceConfig.getAwsAuthenticationConfig().getAwsRegion());
         this.cloudWatchClient = kinesisClientFactory.buildCloudWatchAsyncClient(kinesisLeaseConfig.getLeaseCoordinationTable().getAwsRegion());
-        this.pipelineName = pipelineDescription.getPipelineName();
-        this.applicationName = pipelineName;
+        if (kinesisLeaseConfig.getPipelineIdentifier().isEmpty()) {
+            this.applicationName = pipelineDescription.getPipelineName();
+        } else {
+            this.applicationName = kinesisLeaseConfig.getPipelineIdentifier();
+            }
         this.workerIdentifierGenerator = workerIdentifierGenerator;
         this.executorService = Executors.newFixedThreadPool(1);
         final PluginModel codecConfiguration = kinesisSourceConfig.getCodec();

data-prepper-plugins/kinesis-source/src/test/java/org/opensearch/dataprepper/plugins/kinesis/extension/KinesisLeaseConfigSupplierTest.java

@@ -24,6 +24,7 @@
 @ExtendWith(MockitoExtension.class)
 public class KinesisLeaseConfigSupplierTest {
     private static final String LEASE_COORDINATION_TABLE = "lease-table";
+    private static final String PIPELINE_IDENTIFIER = "sample-kinesis-pipeline-0123";
     @Mock
     KinesisLeaseConfig kinesisLeaseConfig;
 
@@ -36,12 +37,14 @@ private KinesisLeaseConfigSupplier createObjectUnderTest() {
 
     @Test
     void testGetters() {
+        when(kinesisLeaseConfig.getPipelineIdentifier()).thenReturn(PIPELINE_IDENTIFIER);
         when(kinesisLeaseConfig.getLeaseCoordinationTable()).thenReturn(kinesisLeaseCoordinationTableConfig);
         when(kinesisLeaseCoordinationTableConfig.getTableName()).thenReturn(LEASE_COORDINATION_TABLE);
         when(kinesisLeaseCoordinationTableConfig.getRegion()).thenReturn("us-east-1");
         KinesisLeaseConfigSupplier kinesisLeaseConfigSupplier = createObjectUnderTest();
         assertThat(kinesisLeaseConfigSupplier.getKinesisExtensionLeaseConfig().get().getLeaseCoordinationTable().getTableName(), equalTo(LEASE_COORDINATION_TABLE));
         assertThat(kinesisLeaseConfigSupplier.getKinesisExtensionLeaseConfig().get().getLeaseCoordinationTable().getRegion(), equalTo("us-east-1"));
+        assertThat(kinesisLeaseConfigSupplier.getKinesisExtensionLeaseConfig().get().getPipelineIdentifier(), equalTo(PIPELINE_IDENTIFIER));
     }
 
     @Test

data-prepper-plugins/kinesis-source/src/test/java/org/opensearch/dataprepper/plugins/kinesis/extension/KinesisLeaseConfigTest.java

@@ -56,6 +56,8 @@ void testConfigWithTestExtension() throws IOException {
         final KinesisLeaseConfig kinesisLeaseConfig = makeConfig(
                 "src/test/resources/simple_pipeline_with_extensions.yaml");
 
+        assertNotNull(kinesisLeaseConfig.getPipelineIdentifier());
+        assertEquals(kinesisLeaseConfig.getPipelineIdentifier(), "sample-kinesis-pipeline-0123");
         assertNotNull(kinesisLeaseConfig.getLeaseCoordinationTable());
         assertEquals(kinesisLeaseConfig.getLeaseCoordinationTable().getTableName(), "kinesis-pipeline-kcl");
         assertEquals(kinesisLeaseConfig.getLeaseCoordinationTable().getRegion(), "us-east-1");

data-prepper-plugins/kinesis-source/src/test/java/org/opensearch/dataprepper/plugins/kinesis/source/KinesisServiceTest.java

@@ -70,6 +70,7 @@ public class KinesisServiceTest {
     private final String PIPELINE_NAME = "kinesis-pipeline-test";
     private final String streamId = "stream-1";
     private static final String codec_plugin_name = "json";
+    private static final String PIPELINE_IDENTIFIER = "sample-kinesis-pipeline-0123";
 
     private static final Duration CHECKPOINT_INTERVAL = Duration.ofMillis(0);
     private static final int NUMBER_OF_RECORDS_TO_ACCUMULATE = 10;
@@ -149,6 +150,7 @@ void setup() {
         kinesisLeaseConfigSupplier = mock(KinesisLeaseConfigSupplier.class);
         kinesisLeaseConfig = mock(KinesisLeaseConfig.class);
         workerIdentifierGenerator = mock(WorkerIdentifierGenerator.class);
+        when(kinesisLeaseConfig.getPipelineIdentifier()).thenReturn(PIPELINE_IDENTIFIER);
         kinesisLeaseCoordinationTableConfig = mock(KinesisLeaseCoordinationTableConfig.class);
         when(kinesisLeaseConfig.getLeaseCoordinationTable()).thenReturn(kinesisLeaseCoordinationTableConfig);
         when(kinesisLeaseCoordinationTableConfig.getTableName()).thenReturn("kinesis-lease-table");

data-prepper-plugins/kinesis-source/src/test/java/org/opensearch/dataprepper/plugins/kinesis/source/KinesisSourceTest.java

@@ -53,6 +53,7 @@ public class KinesisSourceTest {
     private final String PIPELINE_NAME = "kinesis-pipeline-test";
     private final String streamId = "stream-1";
     private static final String codec_plugin_name = "json";
+    private static final String PIPELINE_IDENTIFIER = "sample-kinesis-pipeline-0123";
 
     @Mock
     private PluginMetrics pluginMetrics;
@@ -110,6 +111,7 @@ void setup() {
 
         kinesisLeaseConfigSupplier = mock(KinesisLeaseConfigSupplier.class);
         kinesisLeaseConfig = mock(KinesisLeaseConfig.class);
+        when(kinesisLeaseConfig.getPipelineIdentifier()).thenReturn(PIPELINE_IDENTIFIER);
         kinesisLeaseCoordinationTableConfig = mock(KinesisLeaseCoordinationTableConfig.class);
         when(kinesisLeaseConfig.getLeaseCoordinationTable()).thenReturn(kinesisLeaseCoordinationTableConfig);
         when(kinesisLeaseCoordinationTableConfig.getTableName()).thenReturn("table-name");

data-prepper-plugins/kinesis-source/src/test/resources/simple_pipeline_with_extensions.yaml

@@ -1,5 +1,6 @@
 extensions:
   kinesis:
+    pipeline_identifier: "sample-kinesis-pipeline-0123"
     lease_coordination:
       table_name: "kinesis-pipeline-kcl"
       region: "us-east-1"