Fix the lease coordination table permissions

data-prepper-plugins/aws-plugin-api/src/main/java/org/opensearch/dataprepper/aws/api/AwsCredentialsOptions.java

@@ -16,23 +16,28 @@
  */
 public class AwsCredentialsOptions {
     private static final AwsCredentialsOptions DEFAULT_OPTIONS = new AwsCredentialsOptions();
+    private static final AwsCredentialsOptions DEFAULT_OPTIONS_WITH_DEFAULT_CREDS =
+            AwsCredentialsOptions.builder().withUseDefaultCredentials(true).build();
     private final String stsRoleArn;
     private final String stsExternalId;
     private final Region region;
     private final Map<String, String> stsHeaderOverrides;
+    private final boolean useDefaultCredentials;
 
     private AwsCredentialsOptions(final Builder builder) {
         this.stsRoleArn = builder.stsRoleArn;
         this.stsExternalId = builder.stsExternalId;
         this.region = builder.region;
         this.stsHeaderOverrides = builder.stsHeaderOverrides != null ? new HashMap<>(builder.stsHeaderOverrides) : Collections.emptyMap();
+        this.useDefaultCredentials = builder.useDefaultCredentials;
     }
 
     private AwsCredentialsOptions() {
         this.stsRoleArn = null;
         this.stsExternalId = null;
         this.region = null;
         this.stsHeaderOverrides = Collections.emptyMap();
+        this.useDefaultCredentials = false;
     }
 
     /**
@@ -49,6 +54,10 @@ public static AwsCredentialsOptions defaultOptions() {
         return DEFAULT_OPTIONS;
     }
 
+    public static AwsCredentialsOptions defaultOptionsWithDefaultCreds() {
+        return DEFAULT_OPTIONS_WITH_DEFAULT_CREDS;
+    }
+
     public String getStsRoleArn() {
         return stsRoleArn;
     }
@@ -65,6 +74,10 @@ public Map<String, String> getStsHeaderOverrides() {
         return stsHeaderOverrides;
     }
 
+    public boolean isUseDefaultCredentials() {
+        return useDefaultCredentials;
+    }
+
     /**
      * Builder class for {@link AwsCredentialsOptions}.
      */
@@ -73,6 +86,7 @@ public static class Builder {
         private String stsExternalId;
         private Region region;
         private Map<String, String> stsHeaderOverrides = Collections.emptyMap();
+        private boolean useDefaultCredentials = false;
 
         /**
          * Sets the STS role ARN to use.
@@ -122,6 +136,17 @@ public Builder withStsHeaderOverrides(final Map<String, String> stsHeaderOverrid
             return this;
         }
 
+        /**
+         * Configures whether to use default credentials.
+         *
+         * @param useDefaultCredentials
+         * @return The {@link Builder} for continuing to build
+         */
+        public Builder withUseDefaultCredentials(final boolean useDefaultCredentials) {
+            this.useDefaultCredentials = useDefaultCredentials;
+            return this;
+        }
+
         /**
          * Builds the {@link AwsCredentialsOptions}.
          *

data-prepper-plugins/aws-plugin-api/src/test/java/org/opensearch/dataprepper/aws/api/AwsCredentialsOptionsTest.java

@@ -19,6 +19,7 @@
 import static org.hamcrest.CoreMatchers.nullValue;
 import static org.hamcrest.CoreMatchers.sameInstance;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
 class AwsCredentialsOptionsTest {
     @Test
@@ -150,4 +151,20 @@ void defaultOptions_returns_same_instance_on_multiple_calls() {
         assertThat(AwsCredentialsOptions.defaultOptions(),
                 sameInstance(AwsCredentialsOptions.defaultOptions()));
     }
+
+
+    @Test
+    void with_DefaultRole() {
+        final AwsCredentialsOptions awsCredentialsOptions = AwsCredentialsOptions.defaultOptionsWithDefaultCreds();
+
+        assertThat(awsCredentialsOptions, notNullValue());
+        assertThat(awsCredentialsOptions.getStsRoleArn(), nullValue());
+        assertTrue(awsCredentialsOptions.isUseDefaultCredentials());
+    }
+
+    @Test
+    void defaultCredentialsOptions_returns_same_instance_on_multiple_calls() {
+        assertThat(AwsCredentialsOptions.defaultOptionsWithDefaultCreds(),
+                sameInstance(AwsCredentialsOptions.defaultOptionsWithDefaultCreds()));
+    }
 }

data-prepper-plugins/aws-plugin/src/main/java/org/opensearch/dataprepper/plugins/aws/CredentialsProviderFactory.java

@@ -50,6 +50,10 @@ Region getDefaultRegion() {
     AwsCredentialsProvider providerFromOptions(final AwsCredentialsOptions credentialsOptions) {
         Objects.requireNonNull(credentialsOptions);
 
+        if (credentialsOptions.isUseDefaultCredentials()) {
+            return DefaultCredentialsProvider.create();
+        }
+
         if(credentialsOptions.getStsRoleArn() != null || defaultStsConfiguration.getAwsStsRoleArn() != null) {
             return createStsCredentials(credentialsOptions);
         }

data-prepper-plugins/aws-plugin/src/test/java/org/opensearch/dataprepper/plugins/aws/AwsPluginIT.java

@@ -157,6 +157,42 @@ void test_AwsPlugin_without_STS_role_and_with_default_role_uses_default_role() {
         assertThat(awsCredentialsProvider2, sameInstance(awsCredentialsProvider1));
     }
 
+    @Test
+    void test_AwsPlugin_without_STS_role_and_without_default_role_uses_default_role() {
+
+        createObjectUnderTest().apply(extensionPoints);
+
+        final ArgumentCaptor<ExtensionProvider<AwsCredentialsSupplier>> extensionProviderArgumentCaptor = ArgumentCaptor.forClass(ExtensionProvider.class);
+        verify(extensionPoints).addExtensionProvider(extensionProviderArgumentCaptor.capture());
+
+        final ExtensionProvider<AwsCredentialsSupplier> extensionProvider = extensionProviderArgumentCaptor.getValue();
+
+        final Optional<AwsCredentialsSupplier> optionalSupplier = extensionProvider.provideInstance(context);
+        assertThat(optionalSupplier, notNullValue());
+        assertThat(optionalSupplier.isPresent(), equalTo(true));
+
+        final AwsCredentialsSupplier awsCredentialsSupplier = optionalSupplier.get();
+
+        final AwsCredentialsOptions awsCredentialsOptions1 = AwsCredentialsOptions.builder()
+                .withRegion(Region.US_EAST_1)
+                .withUseDefaultCredentials(true)
+                .build();
+
+        final AwsCredentialsProvider awsCredentialsProvider1 = awsCredentialsSupplier.getProvider(awsCredentialsOptions1);
+
+        assertThat(awsCredentialsProvider1, instanceOf(DefaultCredentialsProvider.class));
+
+        final AwsCredentialsOptions awsCredentialsOptions2 = AwsCredentialsOptions.builder()
+                .withRegion(Region.US_EAST_1)
+                .withUseDefaultCredentials(true)
+                .build();
+
+        final AwsCredentialsProvider awsCredentialsProvider2 = awsCredentialsSupplier.getProvider(awsCredentialsOptions2);
+
+        assertThat(awsCredentialsProvider2, instanceOf(DefaultCredentialsProvider.class));
+        assertThat(awsCredentialsProvider2, sameInstance(awsCredentialsProvider1));
+    }
+
     private String createStsRole() {
         return String.format("arn:aws:iam::123456789012:role/%s", UUID.randomUUID());
     }

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

@@ -33,7 +33,8 @@ public KinesisClientFactory(final AwsCredentialsSupplier awsCredentialsSupplier,
                 .withStsExternalId(awsAuthenticationConfig.getAwsStsExternalId())
                 .withStsHeaderOverrides(awsAuthenticationConfig.getAwsStsHeaderOverrides())
                 .build());
-        defaultCredentialsProvider = awsCredentialsSupplier.getProvider(AwsCredentialsOptions.defaultOptions());
+        defaultCredentialsProvider = awsCredentialsSupplier.getProvider(
+                AwsCredentialsOptions.defaultOptionsWithDefaultCreds());
         this.awsAuthenticationConfig = awsAuthenticationConfig;
     }