[Test] Add test for AWS SDKv1 swallowing exception at IndexputStream close time (#123505)

This change adds a unit test to demonstrate a specific 
behavior of the AWS SDKv1, which closes the InputStream 
used to upload a blob only after the HTTP request has 
been sent (this is to accomodate for retries). The SDK 
then swallows any exception thrown when closing the 
InputStream which has the effect to hide any potential 
CorruptIndexException that could have been detected 
at that time.

Relates ES-10931

Author: tlrx

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

@@ -21,6 +21,8 @@
 import com.sun.net.httpserver.HttpHandler;
 
 import org.apache.http.HttpStatus;
+import org.apache.lucene.index.CorruptIndexException;
+import org.apache.lucene.store.AlreadyClosedException;
 import org.elasticsearch.ExceptionsHelper;
 import org.elasticsearch.cluster.metadata.RepositoryMetadata;
 import org.elasticsearch.common.BackoffPolicy;
@@ -78,6 +80,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.IntConsumer;
 
 import static org.elasticsearch.repositories.blobstore.BlobStoreTestUtil.randomNonDataPurpose;
@@ -332,6 +335,57 @@ public void testWriteBlobWithReadTimeouts() {
         assertThat(exception.getCause().getCause().getMessage().toLowerCase(Locale.ROOT), containsString("read timed out"));
     }
 
+    /**
+     * This test shows that the AWS SDKv1 defers the closing of the InputStream used to upload a blob after the HTTP request has been sent
+     * to S3, swallowing any exception thrown at closing time.
+     */
+    public void testWriteBlobWithExceptionThrownAtClosingTime() throws Exception {
+        var maxRetries = randomInt(3);
+        var blobLength = randomIntBetween(1, 4096 * 3);
+        var blobName = getTestName().toLowerCase(Locale.ROOT);
+        var blobContainer = createBlobContainer(maxRetries, null, true, null);
+
+        var uploadedBytes = new AtomicReference<BytesReference>();
+        httpServer.createContext(downloadStorageEndpoint(blobContainer, blobName), exchange -> {
+            var requestComponents = S3HttpHandler.parseRequestComponents(S3HttpHandler.getRawRequestString(exchange));
+            if ("PUT".equals(requestComponents.method()) && requestComponents.query().isEmpty()) {
+                var body = Streams.readFully(exchange.getRequestBody());
+                if (uploadedBytes.compareAndSet(null, body)) {
+                    exchange.sendResponseHeaders(HttpStatus.SC_OK, -1);
+                    exchange.close();
+                    return;
+                }
+            }
+            exchange.sendResponseHeaders(HttpStatus.SC_BAD_REQUEST, -1);
+            exchange.close();
+        });
+
+        final byte[] bytes = randomByteArrayOfLength(blobLength);
+
+        var exceptionThrown = new AtomicBoolean();
+        blobContainer.writeBlobAtomic(randomPurpose(), blobName, new FilterInputStream(new ByteArrayInputStream(bytes)) {
+            @Override
+            public void close() throws IOException {
+                if (exceptionThrown.compareAndSet(false, true)) {
+                    switch (randomInt(3)) {
+                        case 0:
+                            throw new CorruptIndexException("simulated", blobName);
+                        case 1:
+                            throw new AlreadyClosedException("simulated");
+                        case 2:
+                            throw new RuntimeException("simulated");
+                        case 3:
+                        default:
+                            throw new IOException("simulated");
+                    }
+                }
+            }
+        }, blobLength, true);
+
+        assertThat(exceptionThrown.get(), is(true));
+        assertArrayEquals(bytes, BytesReference.toBytes(uploadedBytes.get()));
+    }
+
     public void testWriteLargeBlob() throws Exception {
         final boolean useTimeout = rarely();
         final TimeValue readTimeout = useTimeout ? TimeValue.timeValueMillis(randomIntBetween(100, 500)) : null;