(Temporary) Add region override for aws config (opensearch-project#6926)

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

Author: kkondaka

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

@@ -62,7 +62,7 @@ public Map<String, String> getAwsStsHeaderOverrides() {
             "Use either a named configuration reference or inline credentials, not both.")
     public boolean isValidConfiguration() {
         if (configuration != null) {
-            return awsRegion == null && awsStsRoleArn == null && awsStsHeaderOverrides == null && awsStsExternalId == null;
+            return awsStsRoleArn == null && awsStsHeaderOverrides == null && awsStsExternalId == null;
         }
         return true;
     }

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

@@ -73,10 +73,10 @@ void isValidConfiguration_returns_true_when_only_configuration_is_set() throws N
     }
 
     @Test
-    void isValidConfiguration_returns_false_when_configuration_and_region_are_set() throws NoSuchFieldException, IllegalAccessException {
+    void isValidConfiguration_returns_true_when_configuration_and_region_are_set() throws NoSuchFieldException, IllegalAccessException {
         reflectivelySetField(awsConfig, "configuration", "my_config");
         reflectivelySetField(awsConfig, "awsRegion", "us-east-1");
-        assertThat(awsConfig.isValidConfiguration(), equalTo(false));
+        assertThat(awsConfig.isValidConfiguration(), equalTo(true));
     }
 
     @Test

data-prepper-plugins/http-sink/src/main/java/org/opensearch/dataprepper/plugins/sink/http/AwsAuthenticationDecorator.java

@@ -45,7 +45,7 @@ public AwsAuthenticationDecorator(@Nonnull final AwsCredentialsSupplier awsCrede
                                       @Nonnull final String serviceName) {
         this.serviceName = serviceName;
         if (awsConfig != null && awsConfig.getConfiguration() != null) {
-            this.region = awsCredentialsSupplier.getDefaultRegion().orElse(null);
+            this.region = resolveRegion(awsConfig.getAwsRegion(), awsCredentialsSupplier);
             this.credentialsProvider = awsCredentialsSupplier.getProvider(awsConfig.getConfiguration());
         } else if (awsConfig != null) {
             this.region = awsConfig.getAwsRegion();
@@ -57,6 +57,15 @@ public AwsAuthenticationDecorator(@Nonnull final AwsCredentialsSupplier awsCrede
         }
     }
 
+    /**
+     * Resolves region using: 1) use the region provided 2) if no region provided, use the default
+     */
+    private static Region resolveRegion(final Region awsRegion, final AwsCredentialsSupplier awsCredentialsSupplier) {
+        if (awsRegion != null)
+            return awsRegion;
+        return awsCredentialsSupplier.getDefaultRegion().orElseGet(null);
+    }
+
     private static AwsCredentialsOptions convertToCredentialOptions(final AwsConfig awsConfig) {
         return AwsCredentialsOptions.builder()
                 .withRegion(awsConfig.getAwsRegion())

data-prepper-plugins/http-sink/src/test/java/org/opensearch/dataprepper/plugins/sink/http/AwsAuthenticationDecoratorTest.java

@@ -220,4 +220,78 @@ void test_constructor_with_named_configuration() {
         final AwsAuthenticationDecorator decorator = new AwsAuthenticationDecorator(awsCredentialsSupplier, awsConfig, TEST_SERVICE_NAME);
         assertThat(decorator, notNullValue());
     }
+    @Test
+    void test_constructor_with_named_configuration_resolves_region_from_supplier_default() {
+        when(awsConfig.getConfiguration()).thenReturn("my_named_config");
+        when(awsConfig.getAwsRegion()).thenReturn(null);
+        when(awsCredentialsSupplier.getDefaultRegion()).thenReturn(java.util.Optional.of(Region.EU_WEST_1));
+        when(awsCredentialsSupplier.getProvider("my_named_config")).thenReturn(awsCredentialsProvider);
+
+        final AwsAuthenticationDecorator decorator = new AwsAuthenticationDecorator(awsCredentialsSupplier, awsConfig, TEST_SERVICE_NAME);
+
+        final HttpRequest request = decorator.buildRequest(TEST_URL, TEST_PAYLOAD, null);
+        final String authHeader = request.headers().get("Authorization");
+        assertTrue(authHeader.contains("eu-west-1"), "Should use region from supplier default when awsConfig.getAwsRegion() is null");
+    }
+
+    @Test
+    void test_constructor_with_named_configuration_uses_explicit_region_when_set() {
+        when(awsConfig.getConfiguration()).thenReturn("my_named_config");
+        when(awsConfig.getAwsRegion()).thenReturn(Region.AP_SOUTHEAST_1);
+        when(awsCredentialsSupplier.getProvider("my_named_config")).thenReturn(awsCredentialsProvider);
+
+        final AwsAuthenticationDecorator decorator = new AwsAuthenticationDecorator(awsCredentialsSupplier, awsConfig, TEST_SERVICE_NAME);
+
+        final HttpRequest request = decorator.buildRequest(TEST_URL, TEST_PAYLOAD, null);
+        final String authHeader = request.headers().get("Authorization");
+        assertTrue(authHeader.contains("ap-southeast-1"), "Should use explicit region from awsConfig when set");
+    }
+
+    @Test
+    void test_constructor_with_named_configuration_calls_getProvider_with_config_name() {
+        final String configName = "ecs_task_role";
+        when(awsConfig.getConfiguration()).thenReturn(configName);
+        when(awsConfig.getAwsRegion()).thenReturn(TEST_REGION);
+        when(awsCredentialsSupplier.getProvider(configName)).thenReturn(awsCredentialsProvider);
+
+        new AwsAuthenticationDecorator(awsCredentialsSupplier, awsConfig, TEST_SERVICE_NAME);
+
+        verify(awsCredentialsSupplier).getProvider(configName);
+    }
+
+    @Test
+    void test_constructor_with_null_config_uses_default_credentials_provider() {
+        when(awsCredentialsSupplier.getDefaultRegion()).thenReturn(java.util.Optional.of(TEST_REGION));
+        when(awsCredentialsSupplier.getProvider(any(AwsCredentialsOptions.class))).thenReturn(awsCredentialsProvider);
+
+        final AwsAuthenticationDecorator decorator = new AwsAuthenticationDecorator(awsCredentialsSupplier, null, TEST_SERVICE_NAME);
+
+        final ArgumentCaptor<AwsCredentialsOptions> captor = ArgumentCaptor.forClass(AwsCredentialsOptions.class);
+        verify(awsCredentialsSupplier).getProvider(captor.capture());
+        assertTrue(captor.getValue().isUseDefaultCredentialsProvider(),
+                "Should use default credentials provider when awsConfig is null");
+    }
+
+    @Test
+    void test_constructor_with_null_config_resolves_region_from_supplier_default() {
+        when(awsCredentialsSupplier.getDefaultRegion()).thenReturn(java.util.Optional.of(Region.US_WEST_2));
+        when(awsCredentialsSupplier.getProvider(any(AwsCredentialsOptions.class))).thenReturn(awsCredentialsProvider);
+
+        final AwsAuthenticationDecorator decorator = new AwsAuthenticationDecorator(awsCredentialsSupplier, null, TEST_SERVICE_NAME);
+
+        final HttpRequest request = decorator.buildRequest(TEST_URL, TEST_PAYLOAD, null);
+        final String authHeader = request.headers().get("Authorization");
+        assertTrue(authHeader.contains("us-west-2"), "Should use region from supplier default when config is null");
+    }
+
+    @Test
+    void test_constructor_with_inline_config_does_not_call_getProvider_with_string() {
+        // awsConfig has no 'configuration' set (returns null) — should use convertToCredentialOptions path
+        when(awsConfig.getConfiguration()).thenReturn(null);
+
+        createObjectUnderTest();
+
+        verify(awsCredentialsSupplier).getProvider(any(AwsCredentialsOptions.class));
+        verify(awsCredentialsSupplier, org.mockito.Mockito.never()).getProvider(org.mockito.ArgumentMatchers.anyString());
+    }
 }