Must synchronize entire complete-upload process

Author: DaveCTurner

test/fixtures/s3-fixture/src/main/java/fixture/s3/S3HttpHandler.java

@@ -52,6 +52,7 @@
 import static org.elasticsearch.test.fixture.HttpHeaderParser.parseRangeHeader;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
 import static org.w3c.dom.Node.ELEMENT_NODE;
 
 /**
@@ -187,39 +188,42 @@ public void handle(final HttpExchange exchange) throws IOException {
                 }
 
             } else if (request.isCompleteMultipartUploadRequest()) {
-                final var upload = removeUpload(request.getQueryParamOnce("uploadId"));
-                if (upload == null) {
-                    if (Randomness.get().nextBoolean()) {
-                        exchange.sendResponseHeaders(RestStatus.NOT_FOUND.getStatus(), -1);
+                final byte[] responseBody;
+                synchronized (uploads) {
+                    final var upload = removeUpload(request.getQueryParamOnce("uploadId"));
+                    if (upload == null) {
+                        if (Randomness.get().nextBoolean()) {
+                            responseBody = null;
+                        } else {
+                            responseBody = """
+                                <?xml version="1.0" encoding="UTF-8"?>
+                                <Error>
+                                <Code>NoSuchUpload</Code>
+                                <Message>No such upload</Message>
+                                <RequestId>test-request-id</RequestId>
+                                <HostId>test-host-id</HostId>
+                                </Error>""".getBytes(StandardCharsets.UTF_8);
+                        }
                     } else {
-                        byte[] response = ("""
-                            <?xml version="1.0" encoding="UTF-8"?>
-                            <Error>
-                            <Code>NoSuchUpload</Code>
-                            <Message>No such upload</Message>
-                            <RequestId>test-request-id</RequestId>
-                            <HostId>test-host-id</HostId>
-                            </Error>""").getBytes(StandardCharsets.UTF_8);
-                        exchange.getResponseHeaders().add("Content-Type", "application/xml");
-                        exchange.sendResponseHeaders(RestStatus.OK.getStatus(), response.length);
-                        exchange.getResponseBody().write(response);
+                        final var blobContents = upload.complete(extractPartEtags(Streams.readFully(exchange.getRequestBody())));
+                        blobs.put(request.path(), blobContents);
+                        responseBody = ("<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n"
+                            + "<CompleteMultipartUploadResult>\n"
+                            + "<Bucket>"
+                            + bucket
+                            + "</Bucket>\n"
+                            + "<Key>"
+                            + request.path()
+                            + "</Key>\n"
+                            + "</CompleteMultipartUploadResult>").getBytes(StandardCharsets.UTF_8);
                     }
+                }
+                if (responseBody == null) {
+                    exchange.sendResponseHeaders(RestStatus.NOT_FOUND.getStatus(), -1);
                 } else {
-                    final var blobContents = upload.complete(extractPartEtags(Streams.readFully(exchange.getRequestBody())));
-                    blobs.put(request.path(), blobContents);
-
-                    byte[] response = ("<?xml version=\"1.0\" encoding=\"UTF-8\"?>\n"
-                        + "<CompleteMultipartUploadResult>\n"
-                        + "<Bucket>"
-                        + bucket
-                        + "</Bucket>\n"
-                        + "<Key>"
-                        + request.path()
-                        + "</Key>\n"
-                        + "</CompleteMultipartUploadResult>").getBytes(StandardCharsets.UTF_8);
                     exchange.getResponseHeaders().add("Content-Type", "application/xml");
-                    exchange.sendResponseHeaders(RestStatus.OK.getStatus(), response.length);
-                    exchange.getResponseBody().write(response);
+                    exchange.sendResponseHeaders(RestStatus.OK.getStatus(), responseBody.length);
+                    exchange.getResponseBody().write(responseBody);
                 }
             } else if (request.isAbortMultipartUploadRequest()) {
                 final var upload = removeUpload(request.getQueryParamOnce("uploadId"));
@@ -524,7 +528,7 @@ private static HttpHeaderParser.Range parsePartRange(final HttpExchange exchange
     MultipartUpload putUpload(String path) {
         final var upload = new MultipartUpload(UUIDs.randomBase64UUID(), path);
         synchronized (uploads) {
-            uploads.put(upload.getUploadId(), upload);
+            assertNull("upload " + upload.getUploadId() + " should not exist", uploads.put(upload.getUploadId(), upload));
             return upload;
         }
     }