Merge branch 'main' into bugfix/osq

Author: benwtrent

muted-tests.yml

@@ -309,9 +309,6 @@ tests:
 - class: org.elasticsearch.xpack.core.common.notifications.AbstractAuditorTests
   method: testRecreateTemplateWhenDeleted
   issue: https://github.com/elastic/elasticsearch/issues/123232
-- class: org.elasticsearch.xpack.downsample.DataStreamLifecycleDownsampleDisruptionIT
-  method: testDataStreamLifecycleDownsampleRollingRestart
-  issue: https://github.com/elastic/elasticsearch/issues/123769
 - class: org.elasticsearch.xpack.test.rest.XPackRestIT
   method: test {p0=ml/start_data_frame_analytics/Test start given dest index is not empty}
   issue: https://github.com/elastic/elasticsearch/issues/125909

test/framework/src/main/java/org/elasticsearch/test/ClusterServiceUtils.java

@@ -273,6 +273,23 @@ public void clusterStateProcessed(ClusterState initialState, ClusterState newSta
      *         completed exceptionally on the scheduler thread that belongs to {@code clusterService}.
      */
     public static SubscribableListener<Void> addTemporaryStateListener(ClusterService clusterService, Predicate<ClusterState> predicate) {
+        return addTemporaryStateListener(clusterService, predicate, ESTestCase.SAFE_AWAIT_TIMEOUT);
+    }
+
+    /**
+     * Creates a {@link ClusterStateListener} which subscribes to the given {@link ClusterService} and waits for it to apply a cluster state
+     * that satisfies {@code predicate}, at which point it unsubscribes itself.
+     *
+     * @return A {@link SubscribableListener} which is completed when the first cluster state matching {@code predicate} is applied by the
+     *         given {@code clusterService}. If the current cluster state already matches {@code predicate} then the returned listener is
+     *         already complete. If no matching cluster state is seen within the provided {@code timeout} then the listener is
+     *         completed exceptionally on the scheduler thread that belongs to {@code clusterService}.
+     */
+    public static SubscribableListener<Void> addTemporaryStateListener(
+        ClusterService clusterService,
+        Predicate<ClusterState> predicate,
+        TimeValue timeout
+    ) {
         final var listener = new SubscribableListener<Void>();
         final ClusterStateListener clusterStateListener = new ClusterStateListener() {
             @Override
@@ -296,7 +313,7 @@ public String toString() {
         if (predicate.test(clusterService.state())) {
             listener.onResponse(null);
         } else {
-            listener.addTimeout(ESTestCase.SAFE_AWAIT_TIMEOUT, clusterService.threadPool(), EsExecutors.DIRECT_EXECUTOR_SERVICE);
+            listener.addTimeout(timeout, clusterService.threadPool(), EsExecutors.DIRECT_EXECUTOR_SERVICE);
         }
         return listener;
     }

x-pack/plugin/blob-cache/src/main/java/org/elasticsearch/blobcache/BlobCacheMetrics.java

@@ -40,6 +40,10 @@ public enum CachePopulationReason {
          * When warming the cache
          */
         Warming,
+        /**
+         * When warming the cache as a result of an incoming request
+         */
+        OnlinePrewarming,
         /**
          * When the data we need is not in the cache
          */

x-pack/plugin/downsample/src/internalClusterTest/java/org/elasticsearch/xpack/downsample/DataStreamLifecycleDownsampleDisruptionIT.java

@@ -114,7 +114,7 @@ private void ensureDownsamplingStatus(String downsampledIndex, IndexMetadata.Dow
                 return true;
             }
             return false;
-        });
+        }, timeout);
         safeAwait(listener, timeout);
     }
 }

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/external/http/retry/BaseResponseHandler.java

@@ -76,13 +76,21 @@ public String getRequestType() {
     }
 
     @Override
-    public void validateResponse(ThrottlerManager throttlerManager, Logger logger, Request request, HttpResult result) {
+    public void validateResponse(
+        ThrottlerManager throttlerManager,
+        Logger logger,
+        Request request,
+        HttpResult result,
+        boolean checkForErrorObject
+    ) {
         checkForFailureStatusCode(request, result);
         checkForEmptyBody(throttlerManager, logger, request, result);
 
-        // When the response is streamed the status code could be 200 but the error object will be set
-        // so we need to check for that specifically
-        checkForErrorObject(request, result);
+        if (checkForErrorObject) {
+            // When the response is streamed the status code could be 200 but the error object will be set
+            // so we need to check for that specifically
+            checkForErrorObject(request, result);
+        }
     }
 
     protected abstract void checkForFailureStatusCode(Request request, HttpResult result);

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/external/http/retry/ResponseHandler.java

@@ -29,9 +29,12 @@ public interface ResponseHandler {
      * @param logger the logger to use for logging
      * @param request the original request
      * @param result the response from the server
+     * @param checkForErrorObject if true, the validation function should check for the presence of an error object even if the status code
+     *                            indicates a success
      * @throws RetryException if the response is invalid
      */
-    void validateResponse(ThrottlerManager throttlerManager, Logger logger, Request request, HttpResult result) throws RetryException;
+    void validateResponse(ThrottlerManager throttlerManager, Logger logger, Request request, HttpResult result, boolean checkForErrorObject)
+        throws RetryException;
 
     /**
      * A method for parsing the response from the server.

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/external/http/retry/RetryingHttpSender.java

@@ -121,7 +121,7 @@ public void tryAction(ActionListener<InferenceServiceResults> listener) {
                         } else {
                             r.readFullResponse(l.delegateFailureAndWrap((ll, httpResult) -> {
                                 try {
-                                    responseHandler.validateResponse(throttlerManager, logger, request, httpResult);
+                                    responseHandler.validateResponse(throttlerManager, logger, request, httpResult, true);
                                     InferenceServiceResults inferenceResults = responseHandler.parseResult(request, httpResult);
                                     ll.onResponse(inferenceResults);
                                 } catch (Exception e) {
@@ -134,7 +134,7 @@ public void tryAction(ActionListener<InferenceServiceResults> listener) {
                 } else {
                     httpClient.send(request.createHttpRequest(), context, retryableListener.delegateFailure((l, r) -> {
                         try {
-                            responseHandler.validateResponse(throttlerManager, logger, request, r);
+                            responseHandler.validateResponse(throttlerManager, logger, request, r, false);
                             InferenceServiceResults inferenceResults = responseHandler.parseResult(request, r);
 
                             l.onResponse(inferenceResults);

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/services/amazonbedrock/response/AmazonBedrockResponseHandler.java

@@ -22,8 +22,13 @@ public boolean canHandleStreamingResponses() {
     }
 
     @Override
-    public final void validateResponse(ThrottlerManager throttlerManager, Logger logger, Request request, HttpResult result)
-        throws RetryException {
+    public final void validateResponse(
+        ThrottlerManager throttlerManager,
+        Logger logger,
+        Request request,
+        HttpResult result,
+        boolean checkForErrorObject
+    ) throws RetryException {
         // do nothing as the AWS SDK will take care of validation for us
     }
 }

x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/services/azureopenai/response/AzureMistralOpenAiExternalResponseHandler.java

@@ -63,8 +63,13 @@ public AzureMistralOpenAiExternalResponseHandler(
     }
 
     @Override
-    public void validateResponse(ThrottlerManager throttlerManager, Logger logger, Request request, HttpResult result)
-        throws RetryException {
+    public void validateResponse(
+        ThrottlerManager throttlerManager,
+        Logger logger,
+        Request request,
+        HttpResult result,
+        boolean checkForErrorObject
+    ) throws RetryException {
         checkForFailureStatusCode(request, result);
         checkForEmptyBody(throttlerManager, logger, request, result);
     }

x-pack/plugin/inference/src/test/java/org/elasticsearch/xpack/inference/external/http/retry/AlwaysRetryingResponseHandler.java

@@ -35,8 +35,14 @@ public AlwaysRetryingResponseHandler(
         this.parseFunction = Objects.requireNonNull(parseFunction);
     }
 
-    public void validateResponse(ThrottlerManager throttlerManager, Logger logger, Request request, HttpResult result)
-        throws RetryException {
+    @Override
+    public void validateResponse(
+        ThrottlerManager throttlerManager,
+        Logger logger,
+        Request request,
+        HttpResult result,
+        boolean checkForErrorObject
+    ) throws RetryException {
         try {
             checkForFailureStatusCode(throttlerManager, logger, request, result);
             checkForEmptyBody(throttlerManager, logger, request, result);