waitForChe
return this;
}
+ /**
+ * Set the timeout for the {@link #setWaitForCheckpoints(Map)} request.
+ */
+ public SearchRequestBuilder setWaitForCheckpointsTimeout(final TimeValue waitForCheckpointsTimeout) {
+ request.setWaitForCheckpointsTimeout(waitForCheckpointsTimeout);
+ return this;
+ }
+
/**
* Specifies what type of requested indices to ignore and wildcard indices expressions.
*
diff --git a/x-pack/plugin/fleet/qa/rest/build.gradle b/x-pack/plugin/fleet/qa/rest/build.gradle
index fda9251c7ef34..dec624bc3cc56 100644
--- a/x-pack/plugin/fleet/qa/rest/build.gradle
+++ b/x-pack/plugin/fleet/qa/rest/build.gradle
@@ -1,8 +1,15 @@
-apply plugin: 'elasticsearch.legacy-yaml-rest-test'
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
-dependencies {
- yamlRestTestImplementation(testArtifact(project(xpackModule('core'))))
-}
+import org.elasticsearch.gradle.internal.info.BuildParams
+
+apply plugin: 'elasticsearch.internal-yaml-rest-test'
+apply plugin: 'elasticsearch.yaml-rest-compat-test'
+apply plugin: 'elasticsearch.internal-test-artifact'
restResources {
restApi {
@@ -10,11 +17,17 @@ restResources {
}
}
-testClusters.configureEach {
- testDistribution = 'DEFAULT'
- setting 'xpack.security.enabled', 'true'
- setting 'xpack.license.self_generated.type', 'trial'
- extraConfigFile 'roles.yml', file('roles.yml')
- user username: 'elastic_admin', password: 'admin-password'
- user username: 'fleet_unprivileged_secrets', password: 'password', role: 'unprivileged_secrets'
+artifacts {
+ restXpackTests(new File(projectDir, "src/yamlRestTest/resources/rest-api-spec/test"))
+}
+
+tasks.named('yamlRestTest') {
+ usesDefaultDistribution()
+}
+tasks.named('yamlRestCompatTest') {
+ usesDefaultDistribution()
+}
+if (BuildParams.inFipsJvm){
+ // This test cluster is using a BASIC license and FIPS 140 mode is not supported in BASIC
+ tasks.named("yamlRestTest").configure{enabled = false }
}
diff --git a/x-pack/plugin/fleet/qa/rest/src/yamlRestTest/java/org/elasticsearch/xpack/fleet/FleetRestIT.java b/x-pack/plugin/fleet/qa/rest/src/yamlRestTest/java/org/elasticsearch/xpack/fleet/FleetRestIT.java
index 202149abf11e1..bc49649bc1139 100644
--- a/x-pack/plugin/fleet/qa/rest/src/yamlRestTest/java/org/elasticsearch/xpack/fleet/FleetRestIT.java
+++ b/x-pack/plugin/fleet/qa/rest/src/yamlRestTest/java/org/elasticsearch/xpack/fleet/FleetRestIT.java
@@ -12,8 +12,12 @@
import org.elasticsearch.common.settings.SecureString;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.util.concurrent.ThreadContext;
+import org.elasticsearch.test.cluster.ElasticsearchCluster;
+import org.elasticsearch.test.cluster.local.distribution.DistributionType;
+import org.elasticsearch.test.cluster.util.resource.Resource;
import org.elasticsearch.test.rest.yaml.ClientYamlTestCandidate;
import org.elasticsearch.test.rest.yaml.ESClientYamlSuiteTestCase;
+import org.junit.ClassRule;
public class FleetRestIT extends ESClientYamlSuiteTestCase {
@@ -21,14 +25,30 @@ public FleetRestIT(final ClientYamlTestCandidate testCandidate) {
super(testCandidate);
}
+ @ClassRule
+ public static ElasticsearchCluster cluster = ElasticsearchCluster.local()
+ .distribution(DistributionType.DEFAULT)
+ .setting("xpack.license.self_generated.type", "basic")
+ .setting("xpack.security.enabled", "true")
+ .rolesFile(Resource.fromClasspath("roles.yml"))
+ .user("elastic_admin", "admin-password", "superuser", true)
+ .user("fleet_unprivileged_secrets", "password", "unprivileged_secrets", true)
+ .build();
+
+ @Override
+ protected String getTestRestCluster() {
+ return cluster.getHttpAddresses();
+ }
+
@Override
protected Settings restClientSettings() {
- String authentication = basicAuthHeaderValue("elastic_admin", new SecureString("admin-password".toCharArray()));
- return Settings.builder().put(super.restClientSettings()).put(ThreadContext.PREFIX + ".Authorization", authentication).build();
+ String token = basicAuthHeaderValue("elastic_admin", new SecureString("admin-password".toCharArray()));
+ return Settings.builder().put(ThreadContext.PREFIX + ".Authorization", token).build();
}
@ParametersFactory
public static Iterable parameters() throws Exception {
return ESClientYamlSuiteTestCase.createParameters();
}
+
}
diff --git a/x-pack/plugin/fleet/qa/rest/src/yamlRestTest/resources/rest-api-spec/test/fleet/20_wait_for_checkpoints.yml b/x-pack/plugin/fleet/qa/rest/src/yamlRestTest/resources/rest-api-spec/test/fleet/20_wait_for_checkpoints.yml
index 5610502a65d23..4c168c8feb0cd 100644
--- a/x-pack/plugin/fleet/qa/rest/src/yamlRestTest/resources/rest-api-spec/test/fleet/20_wait_for_checkpoints.yml
+++ b/x-pack/plugin/fleet/qa/rest/src/yamlRestTest/resources/rest-api-spec/test/fleet/20_wait_for_checkpoints.yml
@@ -105,6 +105,7 @@ setup:
index: "test-after-refresh"
allow_partial_search_results: false
wait_for_checkpoints: 2
+ wait_for_checkpoints_timeout: 1m
body: { query: { match_all: {} } }
---
@@ -115,7 +116,7 @@ setup:
body:
- { "allow_partial_search_results": false, wait_for_checkpoints: 1 }
- { query: { match_all: { } } }
- - { "allow_partial_search_results": false, wait_for_checkpoints: 2 }
+ - { "allow_partial_search_results": false, wait_for_checkpoints: 2, wait_for_checkpoints_timeout: 1m }
- { query: { match_all: { } } }
- match: { responses.0._shards.successful: 1 }
@@ -128,7 +129,7 @@ setup:
- {query: { match_all: {} } }
- { "index": "test-alias", "allow_partial_search_results": false, wait_for_checkpoints: 1 }
- { query: { match_all: { } } }
- - {"index": "test-refresh-disabled", "allow_partial_search_results": false, wait_for_checkpoints: 2}
+ - { "index": "test-refresh-disabled", "allow_partial_search_results": false, wait_for_checkpoints: 2, wait_for_checkpoints_timeout: 1m }
- {query: { match_all: {} } }
- match: { responses.0._shards.successful: 1 }
diff --git a/x-pack/plugin/fleet/qa/rest/roles.yml b/x-pack/plugin/fleet/qa/rest/src/yamlRestTest/resources/roles.yml
similarity index 100%
rename from x-pack/plugin/fleet/qa/rest/roles.yml
rename to x-pack/plugin/fleet/qa/rest/src/yamlRestTest/resources/roles.yml
From cca70d7eff1540fad2ff19081aac6f5f82cc68f5 Mon Sep 17 00:00:00 2001
From: David Kyle
Date: Tue, 12 Nov 2024 11:14:06 +0000
Subject: [PATCH 4/7] [ML] Batch the chunks (#115477)
Models running on an ml node have a queue of requests, when that queue
is full new requests are rejected. A large document can chunk into hundreds
of requests and in extreme cases a single large document can overflow the
queue. Avoid this by batches of chunks keeping certain number of requests
in flight.
---
.../ElasticsearchInternalService.java | 103 ++++++++++++---
.../EmbeddingRequestChunkerTests.java | 13 ++
.../ElasticsearchInternalServiceTests.java | 122 ++++++++++++++++--
3 files changed, 205 insertions(+), 33 deletions(-)
diff --git a/x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/services/elasticsearch/ElasticsearchInternalService.java b/x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/services/elasticsearch/ElasticsearchInternalService.java
index 83249266c79ab..fe83acc8574aa 100644
--- a/x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/services/elasticsearch/ElasticsearchInternalService.java
+++ b/x-pack/plugin/inference/src/main/java/org/elasticsearch/xpack/inference/services/elasticsearch/ElasticsearchInternalService.java
@@ -68,6 +68,7 @@
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Stream;
@@ -680,25 +681,13 @@ public void chunkedInfer(
esModel.getConfigurations().getChunkingSettings()
).batchRequestsWithListeners(listener);
- for (var batch : batchedRequests) {
- var inferenceRequest = buildInferenceRequest(
- esModel.mlNodeDeploymentId(),
- EmptyConfigUpdate.INSTANCE,
- batch.batch().inputs(),
- inputType,
- timeout
- );
-
- ActionListener mlResultsListener = batch.listener()
- .delegateFailureAndWrap(
- (l, inferenceResult) -> translateToChunkedResult(model.getTaskType(), inferenceResult.getInferenceResults(), l)
- );
-
- var maybeDeployListener = mlResultsListener.delegateResponse(
- (l, exception) -> maybeStartDeployment(esModel, exception, inferenceRequest, mlResultsListener)
- );
-
- client.execute(InferModelAction.INSTANCE, inferenceRequest, maybeDeployListener);
+ if (batchedRequests.isEmpty()) {
+ listener.onResponse(List.of());
+ } else {
+ // Avoid filling the inference queue by executing the batches in series
+ // Each batch contains up to EMBEDDING_MAX_BATCH_SIZE inference request
+ var sequentialRunner = new BatchIterator(esModel, inputType, timeout, batchedRequests);
+ sequentialRunner.run();
}
} else {
listener.onFailure(notElasticsearchModelException(model));
@@ -1018,6 +1007,82 @@ static TaskType inferenceConfigToTaskType(InferenceConfig config) {
}
}
+ /**
+ * Iterates over the batch executing a limited number requests at a time to avoid
+ * filling the ML node inference queue.
+ *
+ * First, a single request is executed, which can also trigger deploying a model
+ * if necessary. When this request is successfully executed, a callback executes
+ * N requests in parallel next. Each of these requests also has a callback that
+ * executes one more request, so that at all time N requests are in-flight. This
+ * continues until all requests are executed.
+ */
+ class BatchIterator {
+ private static final int NUM_REQUESTS_INFLIGHT = 20; // * batch size = 200
+
+ private final AtomicInteger index = new AtomicInteger();
+ private final ElasticsearchInternalModel esModel;
+ private final List requestAndListeners;
+ private final InputType inputType;
+ private final TimeValue timeout;
+
+ BatchIterator(
+ ElasticsearchInternalModel esModel,
+ InputType inputType,
+ TimeValue timeout,
+ List requestAndListeners
+ ) {
+ this.esModel = esModel;
+ this.requestAndListeners = requestAndListeners;
+ this.inputType = inputType;
+ this.timeout = timeout;
+ }
+
+ void run() {
+ // The first request may deploy the model, and upon completion runs
+ // NUM_REQUESTS_INFLIGHT in parallel.
+ inferenceExecutor.execute(() -> inferBatch(NUM_REQUESTS_INFLIGHT, true));
+ }
+
+ private void inferBatch(int runAfterCount, boolean maybeDeploy) {
+ int batchIndex = index.getAndIncrement();
+ if (batchIndex >= requestAndListeners.size()) {
+ return;
+ }
+ executeRequest(batchIndex, maybeDeploy, () -> {
+ for (int i = 0; i < runAfterCount; i++) {
+ // Subsequent requests may not deploy the model, because the first request
+ // already did so. Upon completion, it runs one more request.
+ inferenceExecutor.execute(() -> inferBatch(1, false));
+ }
+ });
+ }
+
+ private void executeRequest(int batchIndex, boolean maybeDeploy, Runnable runAfter) {
+ EmbeddingRequestChunker.BatchRequestAndListener batch = requestAndListeners.get(batchIndex);
+ var inferenceRequest = buildInferenceRequest(
+ esModel.mlNodeDeploymentId(),
+ EmptyConfigUpdate.INSTANCE,
+ batch.batch().inputs(),
+ inputType,
+ timeout
+ );
+ logger.trace("Executing batch index={}", batchIndex);
+
+ ActionListener listener = batch.listener()
+ .delegateFailureAndWrap(
+ (l, inferenceResult) -> translateToChunkedResult(esModel.getTaskType(), inferenceResult.getInferenceResults(), l)
+ );
+ if (runAfter != null) {
+ listener = ActionListener.runAfter(listener, runAfter);
+ }
+ if (maybeDeploy) {
+ listener = listener.delegateResponse((l, exception) -> maybeStartDeployment(esModel, exception, inferenceRequest, l));
+ }
+ client.execute(InferModelAction.INSTANCE, inferenceRequest, listener);
+ }
+ }
+
public static class Configuration {
public static InferenceServiceConfiguration get() {
return configuration.getOrCompute();
diff --git a/x-pack/plugin/inference/src/test/java/org/elasticsearch/xpack/inference/chunking/EmbeddingRequestChunkerTests.java b/x-pack/plugin/inference/src/test/java/org/elasticsearch/xpack/inference/chunking/EmbeddingRequestChunkerTests.java
index c1be537a6b0a7..4fdf254101d3e 100644
--- a/x-pack/plugin/inference/src/test/java/org/elasticsearch/xpack/inference/chunking/EmbeddingRequestChunkerTests.java
+++ b/x-pack/plugin/inference/src/test/java/org/elasticsearch/xpack/inference/chunking/EmbeddingRequestChunkerTests.java
@@ -24,12 +24,25 @@
import java.util.concurrent.atomic.AtomicReference;
import static org.hamcrest.Matchers.contains;
+import static org.hamcrest.Matchers.empty;
import static org.hamcrest.Matchers.hasSize;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.startsWith;
public class EmbeddingRequestChunkerTests extends ESTestCase {
+ public void testEmptyInput() {
+ var embeddingType = randomFrom(EmbeddingRequestChunker.EmbeddingType.values());
+ var batches = new EmbeddingRequestChunker(List.of(), 100, 100, 10, embeddingType).batchRequestsWithListeners(testListener());
+ assertThat(batches, empty());
+ }
+
+ public void testBlankInput() {
+ var embeddingType = randomFrom(EmbeddingRequestChunker.EmbeddingType.values());
+ var batches = new EmbeddingRequestChunker(List.of(""), 100, 100, 10, embeddingType).batchRequestsWithListeners(testListener());
+ assertThat(batches, hasSize(1));
+ }
+
public void testShortInputsAreSingleBatch() {
String input = "one chunk";
var embeddingType = randomFrom(EmbeddingRequestChunker.EmbeddingType.values());
diff --git a/x-pack/plugin/inference/src/test/java/org/elasticsearch/xpack/inference/services/elasticsearch/ElasticsearchInternalServiceTests.java b/x-pack/plugin/inference/src/test/java/org/elasticsearch/xpack/inference/services/elasticsearch/ElasticsearchInternalServiceTests.java
index 89a27a921cbea..9a4d0dda82238 100644
--- a/x-pack/plugin/inference/src/test/java/org/elasticsearch/xpack/inference/services/elasticsearch/ElasticsearchInternalServiceTests.java
+++ b/x-pack/plugin/inference/src/test/java/org/elasticsearch/xpack/inference/services/elasticsearch/ElasticsearchInternalServiceTests.java
@@ -12,6 +12,7 @@
import org.apache.logging.log4j.Level;
import org.elasticsearch.ElasticsearchStatusException;
import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.LatchedActionListener;
import org.elasticsearch.action.support.PlainActionFuture;
import org.elasticsearch.client.internal.Client;
import org.elasticsearch.cluster.service.ClusterService;
@@ -65,6 +66,7 @@
import org.elasticsearch.xpack.inference.InferencePlugin;
import org.elasticsearch.xpack.inference.chunking.ChunkingSettingsTests;
import org.elasticsearch.xpack.inference.chunking.EmbeddingRequestChunker;
+import org.elasticsearch.xpack.inference.chunking.WordBoundaryChunkingSettings;
import org.elasticsearch.xpack.inference.services.ServiceFields;
import org.junit.After;
import org.junit.Before;
@@ -72,12 +74,14 @@
import org.mockito.Mockito;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
@@ -832,16 +836,16 @@ public void testParsePersistedConfig() {
}
}
- public void testChunkInfer_E5WithNullChunkingSettings() {
+ public void testChunkInfer_E5WithNullChunkingSettings() throws InterruptedException {
testChunkInfer_e5(null);
}
- public void testChunkInfer_E5ChunkingSettingsSet() {
+ public void testChunkInfer_E5ChunkingSettingsSet() throws InterruptedException {
testChunkInfer_e5(ChunkingSettingsTests.createRandomChunkingSettings());
}
@SuppressWarnings("unchecked")
- private void testChunkInfer_e5(ChunkingSettings chunkingSettings) {
+ private void testChunkInfer_e5(ChunkingSettings chunkingSettings) throws InterruptedException {
var mlTrainedModelResults = new ArrayList();
mlTrainedModelResults.add(MlTextEmbeddingResultsTests.createRandomResults());
mlTrainedModelResults.add(MlTextEmbeddingResultsTests.createRandomResults());
@@ -889,6 +893,9 @@ private void testChunkInfer_e5(ChunkingSettings chunkingSettings) {
gotResults.set(true);
}, ESTestCase::fail);
+ var latch = new CountDownLatch(1);
+ var latchedListener = new LatchedActionListener<>(resultsListener, latch);
+
service.chunkedInfer(
model,
null,
@@ -897,22 +904,23 @@ private void testChunkInfer_e5(ChunkingSettings chunkingSettings) {
InputType.SEARCH,
new ChunkingOptions(null, null),
InferenceAction.Request.DEFAULT_TIMEOUT,
- ActionListener.runAfter(resultsListener, () -> terminate(threadPool))
+ latchedListener
);
+ latch.await();
assertTrue("Listener not called", gotResults.get());
}
- public void testChunkInfer_SparseWithNullChunkingSettings() {
+ public void testChunkInfer_SparseWithNullChunkingSettings() throws InterruptedException {
testChunkInfer_Sparse(null);
}
- public void testChunkInfer_SparseWithChunkingSettingsSet() {
+ public void testChunkInfer_SparseWithChunkingSettingsSet() throws InterruptedException {
testChunkInfer_Sparse(ChunkingSettingsTests.createRandomChunkingSettings());
}
@SuppressWarnings("unchecked")
- private void testChunkInfer_Sparse(ChunkingSettings chunkingSettings) {
+ private void testChunkInfer_Sparse(ChunkingSettings chunkingSettings) throws InterruptedException {
var mlTrainedModelResults = new ArrayList();
mlTrainedModelResults.add(TextExpansionResultsTests.createRandomResults());
mlTrainedModelResults.add(TextExpansionResultsTests.createRandomResults());
@@ -936,6 +944,7 @@ private void testChunkInfer_Sparse(ChunkingSettings chunkingSettings) {
var service = createService(client);
var gotResults = new AtomicBoolean();
+
var resultsListener = ActionListener.>wrap(chunkedResponse -> {
assertThat(chunkedResponse, hasSize(2));
assertThat(chunkedResponse.get(0), instanceOf(InferenceChunkedSparseEmbeddingResults.class));
@@ -955,6 +964,9 @@ private void testChunkInfer_Sparse(ChunkingSettings chunkingSettings) {
gotResults.set(true);
}, ESTestCase::fail);
+ var latch = new CountDownLatch(1);
+ var latchedListener = new LatchedActionListener<>(resultsListener, latch);
+
service.chunkedInfer(
model,
null,
@@ -963,22 +975,23 @@ private void testChunkInfer_Sparse(ChunkingSettings chunkingSettings) {
InputType.SEARCH,
new ChunkingOptions(null, null),
InferenceAction.Request.DEFAULT_TIMEOUT,
- ActionListener.runAfter(resultsListener, () -> terminate(threadPool))
+ latchedListener
);
+ latch.await();
assertTrue("Listener not called", gotResults.get());
}
- public void testChunkInfer_ElserWithNullChunkingSettings() {
+ public void testChunkInfer_ElserWithNullChunkingSettings() throws InterruptedException {
testChunkInfer_Elser(null);
}
- public void testChunkInfer_ElserWithChunkingSettingsSet() {
+ public void testChunkInfer_ElserWithChunkingSettingsSet() throws InterruptedException {
testChunkInfer_Elser(ChunkingSettingsTests.createRandomChunkingSettings());
}
@SuppressWarnings("unchecked")
- private void testChunkInfer_Elser(ChunkingSettings chunkingSettings) {
+ private void testChunkInfer_Elser(ChunkingSettings chunkingSettings) throws InterruptedException {
var mlTrainedModelResults = new ArrayList();
mlTrainedModelResults.add(TextExpansionResultsTests.createRandomResults());
mlTrainedModelResults.add(TextExpansionResultsTests.createRandomResults());
@@ -1022,6 +1035,9 @@ private void testChunkInfer_Elser(ChunkingSettings chunkingSettings) {
gotResults.set(true);
}, ESTestCase::fail);
+ var latch = new CountDownLatch(1);
+ var latchedListener = new LatchedActionListener<>(resultsListener, latch);
+
service.chunkedInfer(
model,
null,
@@ -1030,9 +1046,10 @@ private void testChunkInfer_Elser(ChunkingSettings chunkingSettings) {
InputType.SEARCH,
new ChunkingOptions(null, null),
InferenceAction.Request.DEFAULT_TIMEOUT,
- ActionListener.runAfter(resultsListener, () -> terminate(threadPool))
+ latchedListener
);
+ latch.await();
assertTrue("Listener not called", gotResults.get());
}
@@ -1093,7 +1110,7 @@ public void testChunkInferSetsTokenization() {
}
@SuppressWarnings("unchecked")
- public void testChunkInfer_FailsBatch() {
+ public void testChunkInfer_FailsBatch() throws InterruptedException {
var mlTrainedModelResults = new ArrayList();
mlTrainedModelResults.add(MlTextEmbeddingResultsTests.createRandomResults());
mlTrainedModelResults.add(MlTextEmbeddingResultsTests.createRandomResults());
@@ -1129,6 +1146,9 @@ public void testChunkInfer_FailsBatch() {
gotResults.set(true);
}, ESTestCase::fail);
+ var latch = new CountDownLatch(1);
+ var latchedListener = new LatchedActionListener<>(resultsListener, latch);
+
service.chunkedInfer(
model,
null,
@@ -1137,12 +1157,86 @@ public void testChunkInfer_FailsBatch() {
InputType.SEARCH,
new ChunkingOptions(null, null),
InferenceAction.Request.DEFAULT_TIMEOUT,
- ActionListener.runAfter(resultsListener, () -> terminate(threadPool))
+ latchedListener
);
+ latch.await();
assertTrue("Listener not called", gotResults.get());
}
+ @SuppressWarnings("unchecked")
+ public void testChunkingLargeDocument() throws InterruptedException {
+ int numBatches = randomIntBetween(3, 6);
+
+ // how many response objects to return in each batch
+ int[] numResponsesPerBatch = new int[numBatches];
+ for (int i = 0; i < numBatches - 1; i++) {
+ numResponsesPerBatch[i] = ElasticsearchInternalService.EMBEDDING_MAX_BATCH_SIZE;
+ }
+ numResponsesPerBatch[numBatches - 1] = randomIntBetween(1, ElasticsearchInternalService.EMBEDDING_MAX_BATCH_SIZE);
+ int numChunks = Arrays.stream(numResponsesPerBatch).sum();
+
+ // build a doc with enough words to make numChunks of chunks
+ int wordsPerChunk = 10;
+ int numWords = numChunks * wordsPerChunk;
+ var input = "word ".repeat(numWords);
+
+ Client client = mock(Client.class);
+ when(client.threadPool()).thenReturn(threadPool);
+
+ // mock the inference response
+ doAnswer(invocationOnMock -> {
+ var request = (InferModelAction.Request) invocationOnMock.getArguments()[1];
+ var listener = (ActionListener) invocationOnMock.getArguments()[2];
+ var mlTrainedModelResults = new ArrayList();
+ for (int i = 0; i < request.numberOfDocuments(); i++) {
+ mlTrainedModelResults.add(MlTextEmbeddingResultsTests.createRandomResults());
+ }
+ var response = new InferModelAction.Response(mlTrainedModelResults, "foo", true);
+ listener.onResponse(response);
+ return null;
+ }).when(client).execute(same(InferModelAction.INSTANCE), any(InferModelAction.Request.class), any(ActionListener.class));
+
+ var service = createService(client);
+
+ var gotResults = new AtomicBoolean();
+ var resultsListener = ActionListener.>wrap(chunkedResponse -> {
+ assertThat(chunkedResponse, hasSize(1));
+ assertThat(chunkedResponse.get(0), instanceOf(InferenceChunkedTextEmbeddingFloatResults.class));
+ var sparseResults = (InferenceChunkedTextEmbeddingFloatResults) chunkedResponse.get(0);
+ assertThat(sparseResults.chunks(), hasSize(numChunks));
+
+ gotResults.set(true);
+ }, ESTestCase::fail);
+
+ // Create model using the word boundary chunker.
+ var model = new MultilingualE5SmallModel(
+ "foo",
+ TaskType.TEXT_EMBEDDING,
+ "e5",
+ new MultilingualE5SmallInternalServiceSettings(1, 1, "cross-platform", null),
+ new WordBoundaryChunkingSettings(wordsPerChunk, 0)
+ );
+
+ var latch = new CountDownLatch(1);
+ var latchedListener = new LatchedActionListener<>(resultsListener, latch);
+
+ // For the given input we know how many requests will be made
+ service.chunkedInfer(
+ model,
+ null,
+ List.of(input),
+ Map.of(),
+ InputType.SEARCH,
+ new ChunkingOptions(null, null),
+ InferenceAction.Request.DEFAULT_TIMEOUT,
+ latchedListener
+ );
+
+ latch.await();
+ assertTrue("Listener not called with results", gotResults.get());
+ }
+
public void testParsePersistedConfig_Rerank() {
// with task settings
{
From 6303de34e44e32372f9d3d4be68bf6d243d9b110 Mon Sep 17 00:00:00 2001
From: Jan Kuipers <148754765+jan-elastic@users.noreply.github.com>
Date: Tue, 12 Nov 2024 12:58:04 +0100
Subject: [PATCH 5/7] Fix NPE in MlMemoryAutoscalingDecider (#116650)
* Fix NPE in MlMemoryAutoscalingDecider
* Update docs/changelog/116650.yaml
* Update 116650.yaml
* Update docs/changelog/116650.yaml
* better fix
---
docs/changelog/116650.yaml | 5 +++++
.../xpack/ml/autoscaling/MlMemoryAutoscalingCapacity.java | 6 +++++-
.../xpack/ml/autoscaling/MlMemoryAutoscalingDecider.java | 2 +-
3 files changed, 11 insertions(+), 2 deletions(-)
create mode 100644 docs/changelog/116650.yaml
diff --git a/docs/changelog/116650.yaml b/docs/changelog/116650.yaml
new file mode 100644
index 0000000000000..d314a918aede9
--- /dev/null
+++ b/docs/changelog/116650.yaml
@@ -0,0 +1,5 @@
+pr: 116650
+summary: Fix bug in ML autoscaling when some node info is unavailable
+area: Machine Learning
+type: bug
+issues: []
diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/autoscaling/MlMemoryAutoscalingCapacity.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/autoscaling/MlMemoryAutoscalingCapacity.java
index bab7bb52f928f..5a06308a3c8cc 100644
--- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/autoscaling/MlMemoryAutoscalingCapacity.java
+++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/autoscaling/MlMemoryAutoscalingCapacity.java
@@ -17,7 +17,11 @@ public static Builder builder(ByteSizeValue nodeSize, ByteSizeValue tierSize) {
}
public static Builder from(AutoscalingCapacity autoscalingCapacity) {
- return builder(autoscalingCapacity.node().memory(), autoscalingCapacity.total().memory());
+ if (autoscalingCapacity == null) {
+ return builder(null, null);
+ } else {
+ return builder(autoscalingCapacity.node().memory(), autoscalingCapacity.total().memory());
+ }
}
@Override
diff --git a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/autoscaling/MlMemoryAutoscalingDecider.java b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/autoscaling/MlMemoryAutoscalingDecider.java
index dfe0e557f749d..0ff6aece95ab1 100644
--- a/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/autoscaling/MlMemoryAutoscalingDecider.java
+++ b/x-pack/plugin/ml/src/main/java/org/elasticsearch/xpack/ml/autoscaling/MlMemoryAutoscalingDecider.java
@@ -809,7 +809,7 @@ static MlMemoryAutoscalingCapacity ensureScaleDown(
MlMemoryAutoscalingCapacity scaleDownResult,
MlMemoryAutoscalingCapacity currentCapacity
) {
- if (scaleDownResult == null || currentCapacity == null) {
+ if (scaleDownResult == null || currentCapacity == null || currentCapacity.isUndetermined()) {
return null;
}
MlMemoryAutoscalingCapacity newCapacity = MlMemoryAutoscalingCapacity.builder(
From 85b2bab2e38409449a37adc9408902cbf79f8c8f Mon Sep 17 00:00:00 2001
From: elasticsearchmachine
Date: Tue, 12 Nov 2024 12:17:04 +0000
Subject: [PATCH 6/7] Bump versions after 8.15.4 release
---
.buildkite/pipelines/intake.yml | 2 +-
.buildkite/pipelines/periodic-packaging.yml | 6 +++---
.buildkite/pipelines/periodic.yml | 10 +++++-----
.ci/bwcVersions | 2 +-
.ci/snapshotBwcVersions | 2 +-
server/src/main/java/org/elasticsearch/Version.java | 1 +
.../resources/org/elasticsearch/TransportVersions.csv | 1 +
.../org/elasticsearch/index/IndexVersions.csv | 1 +
8 files changed, 14 insertions(+), 11 deletions(-)
diff --git a/.buildkite/pipelines/intake.yml b/.buildkite/pipelines/intake.yml
index 37ea49e3a6d95..167830d3ed8b3 100644
--- a/.buildkite/pipelines/intake.yml
+++ b/.buildkite/pipelines/intake.yml
@@ -56,7 +56,7 @@ steps:
timeout_in_minutes: 300
matrix:
setup:
- BWC_VERSION: ["8.15.4", "8.16.0", "8.17.0", "9.0.0"]
+ BWC_VERSION: ["8.15.5", "8.16.0", "8.17.0", "9.0.0"]
agents:
provider: gcp
image: family/elasticsearch-ubuntu-2004
diff --git a/.buildkite/pipelines/periodic-packaging.yml b/.buildkite/pipelines/periodic-packaging.yml
index 788960c76e150..0f2e70addd684 100644
--- a/.buildkite/pipelines/periodic-packaging.yml
+++ b/.buildkite/pipelines/periodic-packaging.yml
@@ -272,8 +272,8 @@ steps:
env:
BWC_VERSION: 8.14.3
- - label: "{{matrix.image}} / 8.15.4 / packaging-tests-upgrade"
- command: ./.ci/scripts/packaging-test.sh -Dbwc.checkout.align=true destructiveDistroUpgradeTest.v8.15.4
+ - label: "{{matrix.image}} / 8.15.5 / packaging-tests-upgrade"
+ command: ./.ci/scripts/packaging-test.sh -Dbwc.checkout.align=true destructiveDistroUpgradeTest.v8.15.5
timeout_in_minutes: 300
matrix:
setup:
@@ -286,7 +286,7 @@ steps:
machineType: custom-16-32768
buildDirectory: /dev/shm/bk
env:
- BWC_VERSION: 8.15.4
+ BWC_VERSION: 8.15.5
- label: "{{matrix.image}} / 8.16.0 / packaging-tests-upgrade"
command: ./.ci/scripts/packaging-test.sh -Dbwc.checkout.align=true destructiveDistroUpgradeTest.v8.16.0
diff --git a/.buildkite/pipelines/periodic.yml b/.buildkite/pipelines/periodic.yml
index 7b6a6ea72fe83..f68f64332426c 100644
--- a/.buildkite/pipelines/periodic.yml
+++ b/.buildkite/pipelines/periodic.yml
@@ -287,8 +287,8 @@ steps:
- signal_reason: agent_stop
limit: 3
- - label: 8.15.4 / bwc
- command: .ci/scripts/run-gradle.sh -Dbwc.checkout.align=true v8.15.4#bwcTest
+ - label: 8.15.5 / bwc
+ command: .ci/scripts/run-gradle.sh -Dbwc.checkout.align=true v8.15.5#bwcTest
timeout_in_minutes: 300
agents:
provider: gcp
@@ -297,7 +297,7 @@ steps:
buildDirectory: /dev/shm/bk
preemptible: true
env:
- BWC_VERSION: 8.15.4
+ BWC_VERSION: 8.15.5
retry:
automatic:
- exit_status: "-1"
@@ -429,7 +429,7 @@ steps:
setup:
ES_RUNTIME_JAVA:
- openjdk21
- BWC_VERSION: ["8.15.4", "8.16.0", "8.17.0", "9.0.0"]
+ BWC_VERSION: ["8.15.5", "8.16.0", "8.17.0", "9.0.0"]
agents:
provider: gcp
image: family/elasticsearch-ubuntu-2004
@@ -471,7 +471,7 @@ steps:
ES_RUNTIME_JAVA:
- openjdk21
- openjdk23
- BWC_VERSION: ["8.15.4", "8.16.0", "8.17.0", "9.0.0"]
+ BWC_VERSION: ["8.15.5", "8.16.0", "8.17.0", "9.0.0"]
agents:
provider: gcp
image: family/elasticsearch-ubuntu-2004
diff --git a/.ci/bwcVersions b/.ci/bwcVersions
index 2e77631450825..b4a4460ff5a80 100644
--- a/.ci/bwcVersions
+++ b/.ci/bwcVersions
@@ -14,7 +14,7 @@ BWC_VERSION:
- "8.12.2"
- "8.13.4"
- "8.14.3"
- - "8.15.4"
+ - "8.15.5"
- "8.16.0"
- "8.17.0"
- "9.0.0"
diff --git a/.ci/snapshotBwcVersions b/.ci/snapshotBwcVersions
index c6edc709a8ceb..7dad55b653925 100644
--- a/.ci/snapshotBwcVersions
+++ b/.ci/snapshotBwcVersions
@@ -1,5 +1,5 @@
BWC_VERSION:
- - "8.15.4"
+ - "8.15.5"
- "8.16.0"
- "8.17.0"
- "9.0.0"
diff --git a/server/src/main/java/org/elasticsearch/Version.java b/server/src/main/java/org/elasticsearch/Version.java
index 5e4df05c10182..909d733fd3719 100644
--- a/server/src/main/java/org/elasticsearch/Version.java
+++ b/server/src/main/java/org/elasticsearch/Version.java
@@ -187,6 +187,7 @@ public class Version implements VersionId, ToXContentFragment {
public static final Version V_8_15_2 = new Version(8_15_02_99);
public static final Version V_8_15_3 = new Version(8_15_03_99);
public static final Version V_8_15_4 = new Version(8_15_04_99);
+ public static final Version V_8_15_5 = new Version(8_15_05_99);
public static final Version V_8_16_0 = new Version(8_16_00_99);
public static final Version V_8_17_0 = new Version(8_17_00_99);
public static final Version V_9_0_0 = new Version(9_00_00_99);
diff --git a/server/src/main/resources/org/elasticsearch/TransportVersions.csv b/server/src/main/resources/org/elasticsearch/TransportVersions.csv
index b0ef5b780e775..26c518962c19a 100644
--- a/server/src/main/resources/org/elasticsearch/TransportVersions.csv
+++ b/server/src/main/resources/org/elasticsearch/TransportVersions.csv
@@ -131,3 +131,4 @@
8.15.1,8702002
8.15.2,8702003
8.15.3,8702003
+8.15.4,8702003
diff --git a/server/src/main/resources/org/elasticsearch/index/IndexVersions.csv b/server/src/main/resources/org/elasticsearch/index/IndexVersions.csv
index e3681cc975988..6cab0b513ee63 100644
--- a/server/src/main/resources/org/elasticsearch/index/IndexVersions.csv
+++ b/server/src/main/resources/org/elasticsearch/index/IndexVersions.csv
@@ -131,3 +131,4 @@
8.15.1,8512000
8.15.2,8512000
8.15.3,8512000
+8.15.4,8512000
From 6c85934c18e22eca5109e66a66174cae339fa040 Mon Sep 17 00:00:00 2001
From: elasticsearchmachine
Date: Tue, 12 Nov 2024 12:18:50 +0000
Subject: [PATCH 7/7] Prune changelogs after 8.15.4 release
---
docs/changelog/112250.yaml | 5 -----
docs/changelog/113723.yaml | 6 ------
docs/changelog/114407.yaml | 6 ------
docs/changelog/114533.yaml | 5 -----
docs/changelog/114601.yaml | 6 ------
docs/changelog/115181.yaml | 5 -----
docs/changelog/115308.yaml | 6 ------
docs/changelog/115430.yaml | 5 -----
docs/changelog/115459.yaml | 5 -----
docs/changelog/115510.yaml | 6 ------
docs/changelog/115834.yaml | 5 -----
docs/changelog/116031.yaml | 6 ------
docs/changelog/116219.yaml | 6 ------
13 files changed, 72 deletions(-)
delete mode 100644 docs/changelog/112250.yaml
delete mode 100644 docs/changelog/113723.yaml
delete mode 100644 docs/changelog/114407.yaml
delete mode 100644 docs/changelog/114533.yaml
delete mode 100644 docs/changelog/114601.yaml
delete mode 100644 docs/changelog/115181.yaml
delete mode 100644 docs/changelog/115308.yaml
delete mode 100644 docs/changelog/115430.yaml
delete mode 100644 docs/changelog/115459.yaml
delete mode 100644 docs/changelog/115510.yaml
delete mode 100644 docs/changelog/115834.yaml
delete mode 100644 docs/changelog/116031.yaml
delete mode 100644 docs/changelog/116219.yaml
diff --git a/docs/changelog/112250.yaml b/docs/changelog/112250.yaml
deleted file mode 100644
index edbb5667d4b9d..0000000000000
--- a/docs/changelog/112250.yaml
+++ /dev/null
@@ -1,5 +0,0 @@
-pr: 112250
-summary: Do not exclude empty arrays or empty objects in source filtering
-area: Search
-type: bug
-issues: [109668]
diff --git a/docs/changelog/113723.yaml b/docs/changelog/113723.yaml
deleted file mode 100644
index 2cbcf49102719..0000000000000
--- a/docs/changelog/113723.yaml
+++ /dev/null
@@ -1,6 +0,0 @@
-pr: 113723
-summary: Fix max file size check to use `getMaxFileSize`
-area: Infra/Core
-type: bug
-issues:
- - 113705
diff --git a/docs/changelog/114407.yaml b/docs/changelog/114407.yaml
deleted file mode 100644
index 4c1134a9d3834..0000000000000
--- a/docs/changelog/114407.yaml
+++ /dev/null
@@ -1,6 +0,0 @@
-pr: 114407
-summary: Fix synthetic source handling for `bit` type in `dense_vector` field
-area: Search
-type: bug
-issues:
- - 114402
diff --git a/docs/changelog/114533.yaml b/docs/changelog/114533.yaml
deleted file mode 100644
index f45589e8de921..0000000000000
--- a/docs/changelog/114533.yaml
+++ /dev/null
@@ -1,5 +0,0 @@
-pr: 114533
-summary: Fix dim validation for bit `element_type`
-area: Vector Search
-type: bug
-issues: []
diff --git a/docs/changelog/114601.yaml b/docs/changelog/114601.yaml
deleted file mode 100644
index d2f563d62a639..0000000000000
--- a/docs/changelog/114601.yaml
+++ /dev/null
@@ -1,6 +0,0 @@
-pr: 114601
-summary: Support semantic_text in object fields
-area: Vector Search
-type: bug
-issues:
- - 114401
diff --git a/docs/changelog/115181.yaml b/docs/changelog/115181.yaml
deleted file mode 100644
index 65f59d5ed0add..0000000000000
--- a/docs/changelog/115181.yaml
+++ /dev/null
@@ -1,5 +0,0 @@
-pr: 115181
-summary: Always check the parent breaker with zero bytes in `PreallocatedCircuitBreakerService`
-area: Aggregations
-type: bug
-issues: []
diff --git a/docs/changelog/115308.yaml b/docs/changelog/115308.yaml
deleted file mode 100644
index 163f0232a3e58..0000000000000
--- a/docs/changelog/115308.yaml
+++ /dev/null
@@ -1,6 +0,0 @@
-pr: 115308
-summary: "ESQL: Disable pushdown of WHERE past STATS"
-area: ES|QL
-type: bug
-issues:
- - 115281
diff --git a/docs/changelog/115430.yaml b/docs/changelog/115430.yaml
deleted file mode 100644
index c2903f7751012..0000000000000
--- a/docs/changelog/115430.yaml
+++ /dev/null
@@ -1,5 +0,0 @@
-pr: 115430
-summary: Prevent NPE if model assignment is removed while waiting to start
-area: Machine Learning
-type: bug
-issues: []
diff --git a/docs/changelog/115459.yaml b/docs/changelog/115459.yaml
deleted file mode 100644
index b20a8f765c084..0000000000000
--- a/docs/changelog/115459.yaml
+++ /dev/null
@@ -1,5 +0,0 @@
-pr: 115459
-summary: Guard blob store local directory creation with `doPrivileged`
-area: Infra/Core
-type: bug
-issues: []
diff --git a/docs/changelog/115510.yaml b/docs/changelog/115510.yaml
deleted file mode 100644
index 1e71270e18f97..0000000000000
--- a/docs/changelog/115510.yaml
+++ /dev/null
@@ -1,6 +0,0 @@
-pr: 115510
-summary: Fix lingering license warning header in IP filter
-area: License
-type: bug
-issues:
- - 114865
diff --git a/docs/changelog/115834.yaml b/docs/changelog/115834.yaml
deleted file mode 100644
index 91f9e9a4e2e41..0000000000000
--- a/docs/changelog/115834.yaml
+++ /dev/null
@@ -1,5 +0,0 @@
-pr: 115834
-summary: Try to simplify geometries that fail with `TopologyException`
-area: Geo
-type: bug
-issues: []
diff --git a/docs/changelog/116031.yaml b/docs/changelog/116031.yaml
deleted file mode 100644
index e30552bf3b513..0000000000000
--- a/docs/changelog/116031.yaml
+++ /dev/null
@@ -1,6 +0,0 @@
-pr: 116031
-summary: Resolve pipelines from template on lazy rollover write
-area: Data streams
-type: bug
-issues:
- - 112781
diff --git a/docs/changelog/116219.yaml b/docs/changelog/116219.yaml
deleted file mode 100644
index aeeea68570e77..0000000000000
--- a/docs/changelog/116219.yaml
+++ /dev/null
@@ -1,6 +0,0 @@
-pr: 116219
-summary: "[apm-data] Apply lazy rollover on index template creation"
-area: Data streams
-type: bug
-issues:
- - 116230