[Test] Fix S3ServiceTests.testRetryOn403RetryPolicy (#115993)

AWS's default retry condition defines a list of retryable error status.
They are retryable for the custom retryable_403 condition as well. This
PR enhances the test to excercise this logic correctly.

Resolves: #115986

Author: ywangd

modules/repository-s3/src/test/java/org/elasticsearch/repositories/s3/S3ServiceTests.java

@@ -9,6 +9,7 @@
 package org.elasticsearch.repositories.s3;
 
 import com.amazonaws.AmazonWebServiceRequest;
+import com.amazonaws.retry.PredefinedRetryPolicies;
 import com.amazonaws.services.s3.model.AmazonS3Exception;
 
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
@@ -19,6 +20,7 @@
 
 import java.io.IOException;
 
+import static org.hamcrest.CoreMatchers.equalTo;
 import static org.mockito.Mockito.mock;
 
 public class S3ServiceTests extends ESTestCase {
@@ -47,19 +49,33 @@ public void testRetryOn403RetryPolicy() {
         e.setStatusCode(403);
         e.setErrorCode("InvalidAccessKeyId");
 
-        // Retry on 403 invalid access key id
+        // AWS default retry condition does not retry on 403
+        assertFalse(PredefinedRetryPolicies.DEFAULT_RETRY_CONDITION.shouldRetry(mock(AmazonWebServiceRequest.class), e, between(0, 9)));
+
+        // The retryable 403 condition retries on 403 invalid access key id
         assertTrue(
             S3Service.RETRYABLE_403_RETRY_POLICY.getRetryCondition().shouldRetry(mock(AmazonWebServiceRequest.class), e, between(0, 9))
         );
 
-        // Not retry if not 403 or not invalid access key id
         if (randomBoolean()) {
+            // Random for another error status that is not 403
             e.setStatusCode(randomValueOtherThan(403, () -> between(0, 600)));
+            // Retryable 403 condition delegates to the AWS default retry condition. Its result must be consistent with the decision
+            // by the AWS default, e.g. some error status like 429 is retryable by default, the retryable 403 condition respects it.
+            boolean actual = S3Service.RETRYABLE_403_RETRY_POLICY.getRetryCondition()
+                .shouldRetry(mock(AmazonWebServiceRequest.class), e, between(0, 9));
+            boolean expected = PredefinedRetryPolicies.DEFAULT_RETRY_CONDITION.shouldRetry(
+                mock(AmazonWebServiceRequest.class),
+                e,
+                between(0, 9)
+            );
+            assertThat(actual, equalTo(expected));
         } else {
+            // Not retry for 403 with error code that is not invalid access key id
             e.setErrorCode(randomAlphaOfLength(10));
+            assertFalse(
+                S3Service.RETRYABLE_403_RETRY_POLICY.getRetryCondition().shouldRetry(mock(AmazonWebServiceRequest.class), e, between(0, 9))
+            );
         }
-        assertFalse(
-            S3Service.RETRYABLE_403_RETRY_POLICY.getRetryCondition().shouldRetry(mock(AmazonWebServiceRequest.class), e, between(0, 9))
-        );
     }
 }

muted-tests.yml

@@ -160,9 +160,6 @@ tests:
 - class: org.elasticsearch.xpack.test.rest.XPackRestIT
   method: test {p0=ml/inference_crud/Test delete given model referenced by pipeline}
   issue: https://github.com/elastic/elasticsearch/issues/115970
-- class: org.elasticsearch.repositories.s3.S3ServiceTests
-  method: testRetryOn403RetryPolicy
-  issue: https://github.com/elastic/elasticsearch/issues/115986
 - class: org.elasticsearch.search.slice.SearchSliceIT
   method: testPointInTime
   issue: https://github.com/elastic/elasticsearch/issues/115988