Merge branch 'main' into do-not-add-use-case-header-twice

Author: timgrein

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

@@ -14,9 +14,7 @@
 import com.amazonaws.DnsResolver;
 import com.amazonaws.SdkClientException;
 import com.amazonaws.services.s3.AmazonS3ClientBuilder;
-import com.amazonaws.services.s3.internal.MD5DigestCalculatingInputStream;
 import com.amazonaws.services.s3.model.AmazonS3Exception;
-import com.amazonaws.util.Base16;
 import com.sun.net.httpserver.HttpExchange;
 import com.sun.net.httpserver.HttpHandler;
 
@@ -30,7 +28,9 @@
 import org.elasticsearch.common.blobstore.BlobPath;
 import org.elasticsearch.common.blobstore.OperationPurpose;
 import org.elasticsearch.common.blobstore.OptionalBytesReference;
+import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.hash.MessageDigests;
 import org.elasticsearch.common.io.Streams;
 import org.elasticsearch.common.lucene.store.ByteArrayIndexInput;
 import org.elasticsearch.common.lucene.store.InputStreamIndexInput;
@@ -377,13 +377,12 @@ public void testWriteLargeBlob() throws Exception {
                 }
             } else if (s3Request.isUploadPartRequest()) {
                 // upload part request
-                MD5DigestCalculatingInputStream md5 = new MD5DigestCalculatingInputStream(exchange.getRequestBody());
-                BytesReference bytes = Streams.readFully(md5);
+                BytesReference bytes = Streams.readFully(exchange.getRequestBody());
                 assertThat((long) bytes.length(), anyOf(equalTo(lastPartSize), equalTo(bufferSize.getBytes())));
                 assertThat(contentLength, anyOf(equalTo(lastPartSize), equalTo(bufferSize.getBytes())));
 
                 if (countDownUploads.decrementAndGet() % 2 == 0) {
-                    exchange.getResponseHeaders().add("ETag", Base16.encodeAsString(md5.getMd5Digest()));
+                    exchange.getResponseHeaders().add("ETag", getBase16MD5Digest(bytes));
                     exchange.sendResponseHeaders(HttpStatus.SC_OK, -1);
                     exchange.close();
                     return;
@@ -475,12 +474,11 @@ public void testWriteLargeBlobStreaming() throws Exception {
                 }
             } else if (s3Request.isUploadPartRequest()) {
                 // upload part request
-                MD5DigestCalculatingInputStream md5 = new MD5DigestCalculatingInputStream(exchange.getRequestBody());
-                BytesReference bytes = Streams.readFully(md5);
+                BytesReference bytes = Streams.readFully(exchange.getRequestBody());
 
                 if (counterUploads.incrementAndGet() % 2 == 0) {
                     bytesReceived.addAndGet(bytes.length());
-                    exchange.getResponseHeaders().add("ETag", Base16.encodeAsString(md5.getMd5Digest()));
+                    exchange.getResponseHeaders().add("ETag", getBase16MD5Digest(bytes));
                     exchange.sendResponseHeaders(HttpStatus.SC_OK, -1);
                     exchange.close();
                     return;
@@ -1105,6 +1103,21 @@ public void testTrimmedLogAndCappedSuppressedErrorOnMultiObjectDeletionException
         }
     }
 
+    private static String getBase16MD5Digest(BytesReference bytesReference) {
+        return MessageDigests.toHexString(MessageDigests.digest(bytesReference, MessageDigests.md5()));
+    }
+
+    public void testGetBase16MD5Digest() {
+        // from Wikipedia, see also org.elasticsearch.common.hash.MessageDigestsTests.testMd5
+        assertBase16MD5Digest("", "d41d8cd98f00b204e9800998ecf8427e");
+        assertBase16MD5Digest("The quick brown fox jumps over the lazy dog", "9e107d9d372bb6826bd81d3542a419d6");
+        assertBase16MD5Digest("The quick brown fox jumps over the lazy dog.", "e4d909c290d0fb1ca068ffaddf22cbd0");
+    }
+
+    private static void assertBase16MD5Digest(String input, String expectedDigestString) {
+        assertEquals(expectedDigestString, getBase16MD5Digest(new BytesArray(input)));
+    }
+
     @Override
     protected Matcher<Integer> getMaxRetriesMatcher(int maxRetries) {
         // some attempts make meaningful progress and do not count towards the max retry limit

muted-tests.yml

@@ -386,6 +386,15 @@ tests:
 - class: org.elasticsearch.snapshots.SharedClusterSnapshotRestoreIT
   method: testDeletionOfFailingToRecoverIndexShouldStopRestore
   issue: https://github.com/elastic/elasticsearch/issues/126204
+- class: org.elasticsearch.index.engine.ThreadPoolMergeSchedulerTests
+  method: testSchedulerCloseWaitsForRunningMerge
+  issue: https://github.com/elastic/elasticsearch/issues/125236
+- class: org.elasticsearch.entitlement.runtime.policy.PolicyUtilsTests
+  method: testFormatFilesEntitlement
+  issue: https://github.com/elastic/elasticsearch/issues/126176
+- class: org.elasticsearch.xpack.security.SecurityRolesMultiProjectIT
+  method: testUpdatingFileBasedRoleAffectsAllProjects
+  issue: https://github.com/elastic/elasticsearch/issues/126223
 
 # Examples:
 #

x-pack/plugin/inference/qa/inference-service-tests/src/javaRestTest/java/org/elasticsearch/xpack/inference/InferenceCrudIT.java

@@ -299,17 +299,13 @@ public void testUnsupportedStream() throws Exception {
 
         try {
             var events = streamInferOnMockService(modelId, TaskType.SPARSE_EMBEDDING, List.of(randomUUID()), null);
-            assertThat(events.size(), equalTo(2));
+            assertThat(events.size(), equalTo(1));
             events.forEach(event -> {
-                switch (event.name()) {
-                    case EVENT -> assertThat(event.value(), equalToIgnoringCase("error"));
-                    case DATA -> assertThat(
-                        event.value(),
-                        containsString(
-                            "Streaming is not allowed for service [streaming_completion_test_service] and task [sparse_embedding]"
-                        )
-                    );
-                }
+                assertThat(event.type(), equalToIgnoringCase("error"));
+                assertThat(
+                    event.data(),
+                    containsString("Streaming is not allowed for service [streaming_completion_test_service] and task [sparse_embedding]")
+                );
             });
         } finally {
             deleteModel(modelId);
@@ -331,12 +327,10 @@ public void testSupportedStream() throws Exception {
                 input.stream().map(s -> s.toUpperCase(Locale.ROOT)).map(str -> "{\"completion\":[{\"delta\":\"" + str + "\"}]}"),
                 Stream.of("[DONE]")
             ).iterator();
-            assertThat(events.size(), equalTo((input.size() + 1) * 2));
+            assertThat(events.size(), equalTo(input.size() + 1));
             events.forEach(event -> {
-                switch (event.name()) {
-                    case EVENT -> assertThat(event.value(), equalToIgnoringCase("message"));
-                    case DATA -> assertThat(event.value(), equalTo(expectedResponses.next()));
-                }
+                assertThat(event.type(), equalToIgnoringCase("message"));
+                assertThat(event.data(), equalTo(expectedResponses.next()));
             });
         } finally {
             deleteModel(modelId);
@@ -359,12 +353,10 @@ public void testUnifiedCompletionInference() throws Exception {
                 VALIDATE_ELASTIC_PRODUCT_HEADER_CONSUMER
             );
             var expectedResponses = expectedResultsIterator(input);
-            assertThat(events.size(), equalTo((input.size() + 1) * 2));
+            assertThat(events.size(), equalTo(input.size() + 1));
             events.forEach(event -> {
-                switch (event.name()) {
-                    case EVENT -> assertThat(event.value(), equalToIgnoringCase("message"));
-                    case DATA -> assertThat(event.value(), equalTo(expectedResponses.next()));
-                }
+                assertThat(event.type(), equalToIgnoringCase("message"));
+                assertThat(event.data(), equalTo(expectedResponses.next()));
             });
         } finally {
             deleteModel(modelId);

x-pack/plugin/inference/src/internalClusterTest/java/org/elasticsearch/xpack/inference/rest/ServerSentEventsRestActionListenerTests.java

@@ -50,7 +50,6 @@
 import org.elasticsearch.xpack.core.inference.action.InferenceAction;
 import org.elasticsearch.xpack.core.inference.results.XContentFormattedException;
 import org.elasticsearch.xpack.inference.external.response.streaming.ServerSentEvent;
-import org.elasticsearch.xpack.inference.external.response.streaming.ServerSentEventField;
 import org.elasticsearch.xpack.inference.external.response.streaming.ServerSentEventParser;
 
 import java.io.IOException;
@@ -353,9 +352,8 @@ private static class RandomStringCollector {
         private void collect(String str) throws IOException {
             sseParser.parse(str.getBytes(StandardCharsets.UTF_8))
                 .stream()
-                .filter(event -> event.name() == ServerSentEventField.DATA)
-                .filter(ServerSentEvent::hasValue)
-                .map(ServerSentEvent::value)
+                .filter(ServerSentEvent::hasData)
+                .map(ServerSentEvent::data)
                 .forEach(stringsVerified::offer);
         }
     }

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/common/DelegatingProcessor.java

@@ -11,7 +11,6 @@
 import org.apache.logging.log4j.Logger;
 import org.elasticsearch.xcontent.XContentParserConfiguration;
 import org.elasticsearch.xpack.inference.external.response.streaming.ServerSentEvent;
-import org.elasticsearch.xpack.inference.external.response.streaming.ServerSentEventField;
 
 import java.io.IOException;
 import java.util.ArrayDeque;
@@ -40,7 +39,7 @@ public static <ParsedChunk> Deque<ParsedChunk> parseEvent(
     ) throws Exception {
         var results = new ArrayDeque<ParsedChunk>(item.size());
         for (ServerSentEvent event : item) {
-            if (ServerSentEventField.DATA == event.name() && event.hasValue()) {
+            if (event.hasData()) {
                 try {
                     var delta = parseFunction.apply(parserConfig, event);
                     delta.forEachRemaining(results::offer);

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/external/response/streaming/ServerSentEvent.java

@@ -9,24 +9,26 @@
 
 /**
  * Server-Sent Event message: https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation
- * Messages always contain a {@link ServerSentEventField} and a non-null payload value.
- * When the stream is parsed and there is no value associated with a {@link ServerSentEventField}, an empty-string is set as the value.
  */
-public record ServerSentEvent(ServerSentEventField name, String value) {
+public record ServerSentEvent(String type, String data) {
 
     private static final String EMPTY = "";
+    private static final String MESSAGE = "message";
 
-    public ServerSentEvent(ServerSentEventField name) {
-        this(name, EMPTY);
+    public static ServerSentEvent empty() {
+        return new ServerSentEvent(EMPTY, EMPTY);
     }
 
-    // treat null value as an empty string, don't break parsing
-    public ServerSentEvent(ServerSentEventField name, String value) {
-        this.name = name;
-        this.value = value != null ? value : EMPTY;
+    public ServerSentEvent(String data) {
+        this(MESSAGE, data);
     }
 
-    public boolean hasValue() {
-        return value.isBlank() == false;
+    public ServerSentEvent {
+        data = data != null ? data : EMPTY;
+        type = type != null && type.isBlank() == false ? type : MESSAGE;
+    }
+
+    public boolean hasData() {
+        return data.isBlank() == false;
     }
 }