Skip to content

Commit f5bde5f

Browse files
original-brownbearoriginal-brownbear
authored
Merge branch 'main' into inline-into-query-phase-res-consumer
2 parents 7b4192c + af686b2 commit f5bde5f

File tree

15 files changed

+113
-39
lines changed

15 files changed

+113
-39
lines changed

‎modules/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3BlobContainer.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -377,7 +377,7 @@ public void copyBlob(
377377
SocketAccess.doPrivilegedVoid(() -> { clientReference.client().copyObject(copyRequest); });
378378
}
379379
}
380-
} catch (final Exception e) {
380+
} catch (final AmazonClientException e) {
381381
if (e instanceof AmazonServiceException ase && ase.getStatusCode() == RestStatus.NOT_FOUND.getStatus()) {
382382
throw new NoSuchFileException(
383383
"Copy source [" + s3SourceBlobContainer.buildKey(sourceBlobName) + "] not found: " + ase.getMessage()
@@ -618,7 +618,7 @@ private void executeMultipart(
618618
SocketAccess.doPrivilegedVoid(() -> clientReference.client().completeMultipartUpload(complRequest));
619619
}
620620
success = true;
621-
} catch (final Exception e) {
621+
} catch (final AmazonClientException e) {
622622
if (e instanceof AmazonServiceException ase && ase.getStatusCode() == RestStatus.NOT_FOUND.getStatus()) {
623623
throw new NoSuchFileException(blobName, null, e.getMessage());
624624
}

‎muted-tests.yml‎

Lines changed: 0 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -309,9 +309,6 @@ tests:
309309
- class: org.elasticsearch.xpack.core.common.notifications.AbstractAuditorTests
310310
method: testRecreateTemplateWhenDeleted
311311
issue: https://github.com/elastic/elasticsearch/issues/123232
312-
- class: org.elasticsearch.xpack.downsample.DataStreamLifecycleDownsampleDisruptionIT
313-
method: testDataStreamLifecycleDownsampleRollingRestart
314-
issue: https://github.com/elastic/elasticsearch/issues/123769
315312
- class: org.elasticsearch.xpack.test.rest.XPackRestIT
316313
method: test {p0=ml/start_data_frame_analytics/Test start given dest index is not empty}
317314
issue: https://github.com/elastic/elasticsearch/issues/125909
@@ -405,21 +402,12 @@ tests:
405402
- class: org.elasticsearch.xpack.esql.action.EsqlActionIT
406403
method: testQueryOnEmptyDataIndex
407404
issue: https://github.com/elastic/elasticsearch/issues/126580
408-
- class: org.elasticsearch.repositories.blobstore.testkit.analyze.S3RepositoryAnalysisRestIT
409-
method: testRepositoryAnalysis
410-
issue: https://github.com/elastic/elasticsearch/issues/126576
411405
- class: org.elasticsearch.xpack.ilm.TimeSeriesDataStreamsIT
412406
method: testShrinkActionInPolicyWithoutHotPhase
413407
issue: https://github.com/elastic/elasticsearch/issues/126746
414-
- class: org.elasticsearch.repositories.blobstore.testkit.analyze.RepositoryAnalysisFailureIT
415-
method: testFailsOnWriteException
416-
issue: https://github.com/elastic/elasticsearch/issues/126747
417408
- class: org.elasticsearch.xpack.test.rest.XPackRestIT
418409
method: test {p0=transform/transforms_start_stop/Test start/stop/start continuous transform}
419410
issue: https://github.com/elastic/elasticsearch/issues/126755
420-
- class: org.elasticsearch.xpack.esql.qa.mixed.MixedClusterEsqlSpecIT
421-
method: test {keep.StatsBySubFieldSortedByKey SYNC}
422-
issue: https://github.com/elastic/elasticsearch/issues/126765
423411

424412
# Examples:
425413
#

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

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -273,6 +273,23 @@ public void clusterStateProcessed(ClusterState initialState, ClusterState newSta
273273
* completed exceptionally on the scheduler thread that belongs to {@code clusterService}.
274274
*/
275275
public static SubscribableListener<Void> addTemporaryStateListener(ClusterService clusterService, Predicate<ClusterState> predicate) {
276+
return addTemporaryStateListener(clusterService, predicate, ESTestCase.SAFE_AWAIT_TIMEOUT);
277+
}
278+
279+
/**
280+
* Creates a {@link ClusterStateListener} which subscribes to the given {@link ClusterService} and waits for it to apply a cluster state
281+
* that satisfies {@code predicate}, at which point it unsubscribes itself.
282+
*
283+
* @return A {@link SubscribableListener} which is completed when the first cluster state matching {@code predicate} is applied by the
284+
* given {@code clusterService}. If the current cluster state already matches {@code predicate} then the returned listener is
285+
* already complete. If no matching cluster state is seen within the provided {@code timeout} then the listener is
286+
* completed exceptionally on the scheduler thread that belongs to {@code clusterService}.
287+
*/
288+
public static SubscribableListener<Void> addTemporaryStateListener(
289+
ClusterService clusterService,
290+
Predicate<ClusterState> predicate,
291+
TimeValue timeout
292+
) {
276293
final var listener = new SubscribableListener<Void>();
277294
final ClusterStateListener clusterStateListener = new ClusterStateListener() {
278295
@Override
@@ -296,7 +313,7 @@ public String toString() {
296313
if (predicate.test(clusterService.state())) {
297314
listener.onResponse(null);
298315
} else {
299-
listener.addTimeout(ESTestCase.SAFE_AWAIT_TIMEOUT, clusterService.threadPool(), EsExecutors.DIRECT_EXECUTOR_SERVICE);
316+
listener.addTimeout(timeout, clusterService.threadPool(), EsExecutors.DIRECT_EXECUTOR_SERVICE);
300317
}
301318
return listener;
302319
}

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

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,10 @@ public enum CachePopulationReason {
4040
* When warming the cache
4141
*/
4242
Warming,
43+
/**
44+
* When warming the cache as a result of an incoming request
45+
*/
46+
OnlinePrewarming,
4347
/**
4448
* When the data we need is not in the cache
4549
*/

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -114,7 +114,7 @@ private void ensureDownsamplingStatus(String downsampledIndex, IndexMetadata.Dow
114114
return true;
115115
}
116116
return false;
117-
});
117+
}, timeout);
118118
safeAwait(listener, timeout);
119119
}
120120
}

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

Lines changed: 12 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -76,13 +76,21 @@ public String getRequestType() {
7676
}
7777

7878
@Override
79-
public void validateResponse(ThrottlerManager throttlerManager, Logger logger, Request request, HttpResult result) {
79+
public void validateResponse(
80+
ThrottlerManager throttlerManager,
81+
Logger logger,
82+
Request request,
83+
HttpResult result,
84+
boolean checkForErrorObject
85+
) {
8086
checkForFailureStatusCode(request, result);
8187
checkForEmptyBody(throttlerManager, logger, request, result);
8288

83-
// When the response is streamed the status code could be 200 but the error object will be set
84-
// so we need to check for that specifically
85-
checkForErrorObject(request, result);
89+
if (checkForErrorObject) {
90+
// When the response is streamed the status code could be 200 but the error object will be set
91+
// so we need to check for that specifically
92+
checkForErrorObject(request, result);
93+
}
8694
}
8795

8896
protected abstract void checkForFailureStatusCode(Request request, HttpResult result);

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

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -29,9 +29,12 @@ public interface ResponseHandler {
2929
* @param logger the logger to use for logging
3030
* @param request the original request
3131
* @param result the response from the server
32+
* @param checkForErrorObject if true, the validation function should check for the presence of an error object even if the status code
33+
* indicates a success
3234
* @throws RetryException if the response is invalid
3335
*/
34-
void validateResponse(ThrottlerManager throttlerManager, Logger logger, Request request, HttpResult result) throws RetryException;
36+
void validateResponse(ThrottlerManager throttlerManager, Logger logger, Request request, HttpResult result, boolean checkForErrorObject)
37+
throws RetryException;
3538

3639
/**
3740
* 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‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -121,7 +121,7 @@ public void tryAction(ActionListener<InferenceServiceResults> listener) {
121121
} else {
122122
r.readFullResponse(l.delegateFailureAndWrap((ll, httpResult) -> {
123123
try {
124-
responseHandler.validateResponse(throttlerManager, logger, request, httpResult);
124+
responseHandler.validateResponse(throttlerManager, logger, request, httpResult, true);
125125
InferenceServiceResults inferenceResults = responseHandler.parseResult(request, httpResult);
126126
ll.onResponse(inferenceResults);
127127
} catch (Exception e) {
@@ -134,7 +134,7 @@ public void tryAction(ActionListener<InferenceServiceResults> listener) {
134134
} else {
135135
httpClient.send(request.createHttpRequest(), context, retryableListener.delegateFailure((l, r) -> {
136136
try {
137-
responseHandler.validateResponse(throttlerManager, logger, request, r);
137+
responseHandler.validateResponse(throttlerManager, logger, request, r, false);
138138
InferenceServiceResults inferenceResults = responseHandler.parseResult(request, r);
139139

140140
l.onResponse(inferenceResults);

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

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,13 @@ public boolean canHandleStreamingResponses() {
2222
}
2323

2424
@Override
25-
public final void validateResponse(ThrottlerManager throttlerManager, Logger logger, Request request, HttpResult result)
26-
throws RetryException {
25+
public final void validateResponse(
26+
ThrottlerManager throttlerManager,
27+
Logger logger,
28+
Request request,
29+
HttpResult result,
30+
boolean checkForErrorObject
31+
) throws RetryException {
2732
// do nothing as the AWS SDK will take care of validation for us
2833
}
2934
}

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

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -63,8 +63,13 @@ public AzureMistralOpenAiExternalResponseHandler(
6363
}
6464

6565
@Override
66-
public void validateResponse(ThrottlerManager throttlerManager, Logger logger, Request request, HttpResult result)
67-
throws RetryException {
66+
public void validateResponse(
67+
ThrottlerManager throttlerManager,
68+
Logger logger,
69+
Request request,
70+
HttpResult result,
71+
boolean checkForErrorObject
72+
) throws RetryException {
6873
checkForFailureStatusCode(request, result);
6974
checkForEmptyBody(throttlerManager, logger, request, result);
7075
}

0 commit comments

Comments
 (0)