(fix) adding support for considering table-name and table-arn independently just for ddblocal

Author: amdhing

data-prepper-plugins/dynamodb-source/src/main/java/org/opensearch/dataprepper/plugins/source/dynamodb/leader/LeaderScheduler.java

@@ -234,7 +234,10 @@ private void compareAndCreateChildrenPartitions(List<EnhancedSourcePartition> so
      * Once created, the info should not be changed.
      */
     private TableInfo getTableInfo(final TableConfig tableConfig) {
-        final String tableName = tableConfig.getTableArn();
+        final String tableArn = tableConfig.getTableArn();
+        final String tableName = tableArn.contains("/") ?
+                tableArn.substring(tableArn.lastIndexOf('/') + 1) :
+                tableArn;
         DescribeTableResponse describeTableResult;
         try {
             // Need to call describe table to get the Key schema for table

data-prepper-plugins/dynamodb-source/src/test/java/org/opensearch/dataprepper/plugins/source/dynamodb/leader/LeaderSchedulerTest.java

@@ -227,7 +227,7 @@ void test_should_init() throws InterruptedException {
         ArgumentCaptor<DescribeTableRequest> describeTableRequestArgumentCaptor = ArgumentCaptor.forClass(DescribeTableRequest.class);
         verify(dynamoDbClient).describeTable(describeTableRequestArgumentCaptor.capture());
         DescribeTableRequest actualDescribeTableRequest = describeTableRequestArgumentCaptor.getValue();
-        assertThat(actualDescribeTableRequest.tableName(), equalTo(tableArn));
+        assertThat(actualDescribeTableRequest.tableName(), equalTo(tableName));
         // Should check PITR enabled or not
         verify(dynamoDbClient).describeContinuousBackups(any(DescribeContinuousBackupsRequest.class));
         // Acquire the init partition
@@ -259,7 +259,7 @@ void test_PITR_not_enabled_init_should_failed() throws InterruptedException {
         ArgumentCaptor<DescribeTableRequest> describeTableRequestArgumentCaptor = ArgumentCaptor.forClass(DescribeTableRequest.class);
         verify(dynamoDbClient).describeTable(describeTableRequestArgumentCaptor.capture());
         DescribeTableRequest actualDescribeTableRequest = describeTableRequestArgumentCaptor.getValue();
-        assertThat(actualDescribeTableRequest.tableName(), equalTo(tableArn));
+        assertThat(actualDescribeTableRequest.tableName(), equalTo(tableName));
 
         // Should check PITR enabled or not
         verify(dynamoDbClient).describeContinuousBackups(any(DescribeContinuousBackupsRequest.class));
@@ -288,7 +288,7 @@ void test_streaming_not_enabled_init_should_failed() throws InterruptedException
         ArgumentCaptor<DescribeTableRequest> describeTableRequestArgumentCaptor = ArgumentCaptor.forClass(DescribeTableRequest.class);
         verify(dynamoDbClient).describeTable(describeTableRequestArgumentCaptor.capture());
         DescribeTableRequest actualDescribeTableRequest = describeTableRequestArgumentCaptor.getValue();
-        assertThat(actualDescribeTableRequest.tableName(), equalTo(tableArn));
+        assertThat(actualDescribeTableRequest.tableName(), equalTo(tableName));
 
         // Should check PITR enabled or not
         verify(dynamoDbClient).describeContinuousBackups(any(DescribeContinuousBackupsRequest.class));