Skip to content

Batch index creation #144074

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
inespot merged 20 commits into from
Mar 20, 2026
Merged

Batch index creation #144074

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
20 commits
Select commit Hold shift + click to select a range
0010211
Batch create-index tasks
DaveCTurner Feb 15, 2026
3cc912d
Small nits and fixes
inespot Mar 12, 2026
afad4a3
Add small comment
inespot Mar 13, 2026
bcdfc8c
Revert param position move
inespot Mar 13, 2026
4fddbce
Add failure testing + fix unused variables
inespot Mar 13, 2026
0e8d371
More testing + style fixes
inespot Mar 14, 2026
04e9043
Merge branch 'main' into feature/batch-index-creation
inespot Mar 14, 2026
da1a6a2
Rename + small javadoc
inespot Mar 14, 2026
0a237dc
More nits
inespot Mar 14, 2026
e23f2f1
Further strengthen tests
inespot Mar 16, 2026
cf274f7
Merge branch 'main' into feature/batch-index-creation
inespot Mar 16, 2026
3a52952
Tests improvement: no collision, latches & nits
inespot Mar 19, 2026
23eabb5
testCreateIndexBatching supports multi projects
inespot Mar 20, 2026
8a60cc3
Merge branch 'main' into feature/batch-index-creation
inespot Mar 20, 2026
5a74bc6
Revert "testCreateIndexBatching supports multi projects"
inespot Mar 20, 2026
d1e63eb
Test consolidate request build and send
inespot Mar 20, 2026
79a14af
Merge branch 'main' into feature/batch-index-creation
inespot Mar 20, 2026
216bebe
Missed this one
inespot Mar 20, 2026
4128e56
Update docs/changelog/144074.yaml
inespot Mar 20, 2026
1eef9fe
Merge branch 'main' into feature/batch-index-creation
inespot Mar 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
testCreateIndexBatching supports multi projects
  • Loading branch information
@inespot
inespot committed Mar 20, 2026
commit 23eabb54acd8592822d549a084675395c8d0603b
1 change: 1 addition & 0 deletions server/build.gradle
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -90,6 +90,7 @@ dependencies {
internalClusterTestImplementation(project(':modules:reindex'))
internalClusterTestImplementation(project(':modules:mapper-extras'))
internalClusterTestImplementation(project(':modules:data-streams'))
internalClusterTestImplementation(project(':test:external-modules:test-multi-project'))
}

spotless {
Expand Down
242 changes: 180 additions & 62 deletions ...rnalClusterTest/java/org/elasticsearch/cluster/metadata/MetadataCreateIndexServiceIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,25 +14,30 @@
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.LatchedActionListener;
import org.elasticsearch.action.admin.indices.create.CreateIndexClusterStateUpdateRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexRequest;
import org.elasticsearch.action.admin.indices.create.CreateIndexResponse;
import org.elasticsearch.action.admin.indices.create.TransportCreateIndexAction;
import org.elasticsearch.action.admin.indices.get.GetIndexRequest;
import org.elasticsearch.action.admin.indices.get.GetIndexResponse;
import org.elasticsearch.action.support.ActiveShardCount;
import org.elasticsearch.action.support.master.ShardsAcknowledgedResponse;
import org.elasticsearch.cluster.ClusterChangedEvent;
import org.elasticsearch.cluster.ClusterState;
import org.elasticsearch.cluster.ClusterStateListener;
import org.elasticsearch.cluster.ClusterStateUpdateTask;
import org.elasticsearch.cluster.service.ClusterService;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.core.TimeValue;
import org.elasticsearch.indices.InvalidIndexNameException;
import org.elasticsearch.multiproject.TestOnlyMultiProjectPlugin;
import org.elasticsearch.plugins.Plugin;
import org.elasticsearch.test.ESIntegTestCase;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.CyclicBarrier;
Expand All @@ -47,6 +52,38 @@

public class MetadataCreateIndexServiceIT extends ESIntegTestCase {

@Override
protected Collection<Class<? extends Plugin>> nodePlugins() {
return List.of(TestOnlyMultiProjectPlugin.class);
}

@Override
protected Settings nodeSettings(int nodeOrdinal, Settings otherSettings) {
return Settings.builder()
.put(super.nodeSettings(nodeOrdinal, otherSettings))
.put(TestOnlyMultiProjectPlugin.MULTI_PROJECT_ENABLED.getKey(), true)
.build();
}

@Override
protected boolean multiProjectIntegrationTest() {
return true;
}

private enum CreateIndexResult {
SUCCESS,
INVALID_NAME,
INVALID_SETTINGS,
ALREADY_EXISTS
}

private record CreateIndexRequestSpec(
String indexName,
boolean invalidSettings,
CreateIndexResult expectedResult,
ProjectId projectId
) {}

public void testRequestTemplateIsRespected() throws InterruptedException {
/*
* This test passes a template in the CreateIndexClusterStateUpdateRequest, and makes sure that the settings
Expand Down Expand Up @@ -99,66 +136,61 @@ public void onFailure(Exception e) {

public void testCreateIndexBatching() throws Exception {
final var masterClusterService = internalCluster().getCurrentMasterNodeInstance(ClusterService.class);
final var metadataCreateIndexService = internalCluster().getCurrentMasterNodeInstance(MetadataCreateIndexService.class);

enum ExpectedResult {
SUCCESS,
INVALID_NAME,
INVALID_SETTINGS,
ALREADY_EXISTS
final var projects = new HashSet<ProjectId>();
final var projectCount = randomIntBetween(1, 5);
while (projects.size() < projectCount) {
projects.add(randomUniqueProjectId());
}
addProjectsToClusterState(masterClusterService, projects);

record RequestSpec(String indexName, boolean invalidSettings, ExpectedResult expectedResult) {}

final var allIndexNames = new HashSet<String>();
final int totalRequestCount = randomIntBetween(1, 20);
final int validRequestCount = randomIntBetween(0, totalRequestCount);
final int invalidRequestCount = totalRequestCount - validRequestCount;

final var allRequests = new ArrayList<RequestSpec>(totalRequestCount);
final var validIndicesNames = new HashSet<String>();
final var invalidSettingsNames = new HashSet<String>();
final var allIndicesNames = new HashSet<String>();
// The same index name can exist in different projects. Pre-create it in all of them
final var preExistingIndexName = addRandomIndexNameNoCollision(allIndexNames);
addIndexToAllProjects(metadataCreateIndexService, preExistingIndexName, projects);

final var allRequests = new ArrayList<CreateIndexRequestSpec>(totalRequestCount);
final var validIndicesByProject = new HashMap<ProjectId, Set<String>>(projects.size());
for (int i = 0; i < validRequestCount; i++) {
final var indexName = addRandomIndexNameNoCollision(allIndicesNames);
validIndicesNames.add(indexName);
allRequests.add(new RequestSpec(indexName, false, ExpectedResult.SUCCESS));
final var projectId = randomFrom(projects);
final var indexName = addRandomIndexNameNoCollision(allIndexNames);
validIndicesByProject.putIfAbsent(projectId, new HashSet<>());
validIndicesByProject.get(projectId).add(indexName);
allRequests.add(new CreateIndexRequestSpec(indexName, false, CreateIndexResult.SUCCESS, projectId));

@DaveCTurner DaveCTurner Mar 20, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do we need to track all the request-specs in a separate data structure like this? I would expect we can do this with one loop, after the master service is blocked, which constructs each request together with its expected-result listener, and immediately sends it.

@inespot inespot Mar 20, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You are right, and that's much nicer, thanks! Refactored as suggested in d1e63e

}

final var preExistingIndexName = addRandomIndexNameNoCollision(allIndicesNames);
createIndex(preExistingIndexName);

final var invalidSettingsNames = new HashSet<String>();
for (int i = 0; i < invalidRequestCount; i++) {
int failureType = validIndicesNames.isEmpty() ? randomIntBetween(0, 1) : randomIntBetween(0, 2);
switch (failureType) {
case 0 -> allRequests.add(
new RequestSpec(randomIdentifier("INVALID_BECAUSE_UPPER_CASE_"), false, ExpectedResult.INVALID_NAME)
);
final var projectId = randomFrom(projects);
switch (randomIntBetween(0, 2)) {
case 0 -> {
final var indexName = randomIdentifier("INVALID_BECAUSE_UPPER_CASE_");
allRequests.add(new CreateIndexRequestSpec(indexName, false, CreateIndexResult.INVALID_NAME, projectId));
}
case 1 -> {
var indexName = addRandomIndexNameNoCollision(allIndicesNames);
final var indexName = addRandomIndexNameNoCollision(allIndexNames);
invalidSettingsNames.add(indexName);
allRequests.add(new RequestSpec(indexName, true, ExpectedResult.INVALID_SETTINGS));
allRequests.add(new CreateIndexRequestSpec(indexName, true, CreateIndexResult.INVALID_SETTINGS, projectId));
}
default -> allRequests.add(new RequestSpec(preExistingIndexName, false, ExpectedResult.ALREADY_EXISTS));
default -> allRequests.add(
new CreateIndexRequestSpec(preExistingIndexName, false, CreateIndexResult.ALREADY_EXISTS, projectId)
);
}
}
Collections.shuffle(allRequests, random());

final ClusterStateListener listener = event -> {
final var projectMetadata = event.state().metadata().getProject(ProjectId.DEFAULT);
if (projectMetadata == null) {
return;
}
final var createdInState = validIndicesNames.stream().filter(projectMetadata::hasIndex).toList();
assertThat(
"expected either none or all valid indices to appear atomically, but found " + createdInState.size(),
createdInState,
anyOf(hasSize(0), hasSize(validIndicesNames.size()))
);
for (final var indexName : invalidSettingsNames) {
assertFalse("invalid index [" + indexName + "] should never be created", projectMetadata.hasIndex(indexName));
}
};
final ClusterStateListener listener = event -> assertAllIndicesCreatedAtomically(
event,
projects,
validIndicesByProject,
invalidSettingsNames
);
masterClusterService.addListener(listener);

try {
final var barrier = new CyclicBarrier(2);
masterClusterService.submitUnbatchedStateUpdateTask("block", new ClusterStateUpdateTask() {
Expand All @@ -177,25 +209,24 @@ public void onFailure(Exception e) {
safeAwait(barrier);

final var responsesLatch = new CountDownLatch(allRequests.size());
for (var requestSpec : allRequests) {
final var request = new CreateIndexRequest(requestSpec.indexName());
for (final var requestSpec : allRequests) {
final var settingsBuilder = Settings.builder().put("index.routing.allocation.enable", "none");
if (requestSpec.invalidSettings()) {
request.settings(Settings.builder().put("index.version.created", 1));
settingsBuilder.put("index.version.created", 1);
}

final ActionListener<CreateIndexResponse> delegate = switch (requestSpec.expectedResult()) {
case SUCCESS -> assertNoFailureListener(response -> assertTrue(response.isAcknowledged()));
case INVALID_NAME -> assertNoSuccessListener(
e -> assertThat(ExceptionsHelper.unwrapCause(e), instanceOf(InvalidIndexNameException.class))
);
case INVALID_SETTINGS -> assertNoSuccessListener(
e -> assertThat(ExceptionsHelper.unwrapCause(e), instanceOf(IllegalArgumentException.class))
);
case ALREADY_EXISTS -> assertNoSuccessListener(
e -> assertThat(ExceptionsHelper.unwrapCause(e), instanceOf(ResourceAlreadyExistsException.class))
);
};
client().execute(TransportCreateIndexAction.TYPE, request, new LatchedActionListener<>(delegate, responsesLatch));
final var request = new CreateIndexClusterStateUpdateRequest(
"create-index",
requestSpec.projectId(),
requestSpec.indexName(),
randomAlphaOfLength(20)
).settings(settingsBuilder.build()).waitForActiveShards(ActiveShardCount.NONE);
metadataCreateIndexService.createIndex(
TimeValue.THIRTY_SECONDS,
TimeValue.THIRTY_SECONDS,
TimeValue.THIRTY_SECONDS,
request,
new LatchedActionListener<>(listenerForCreateIndexResult(requestSpec.expectedResult()), responsesLatch)
);
}

assertTrue(
Expand All @@ -212,14 +243,101 @@ public void onFailure(Exception e) {
safeAwait(barrier);

assertTrue("timed out waiting for create-index responses", responsesLatch.await(30, TimeUnit.SECONDS));
if (validIndicesNames.isEmpty()) {
if (validIndicesByProject.isEmpty()) {
assertSame("cluster state should not change when all requests failed", masterClusterService.state(), initialState);
}
} finally {
masterClusterService.removeListener(listener);
}
}

private void assertAllIndicesCreatedAtomically(
ClusterChangedEvent event,
Set<ProjectId> projects,
Map<ProjectId, Set<String>> indicesPerProject,
Set<String> invalidIndices
) {
var projectsWithValidIndices = 0;
var allCreatedInProject = 0;
for (var projectId : projects) {
final var projectMetadata = event.state().metadata().getProject(projectId);
final var expectedIndicesInProject = indicesPerProject.getOrDefault(projectId, Set.of());
final var createdInProject = expectedIndicesInProject.stream().filter(projectMetadata::hasIndex).toList();
assertThat(
"expected either none or all valid indices to appear atomically in single project",
createdInProject,
anyOf(hasSize(0), hasSize(expectedIndicesInProject.size()))
);
if (expectedIndicesInProject.isEmpty() == false) {
projectsWithValidIndices += 1;
if (createdInProject.size() == expectedIndicesInProject.size()) {
allCreatedInProject += 1;
}
}
invalidIndices.forEach(
invalid -> assertFalse("invalid index [" + invalid + "] should never be created", projectMetadata.hasIndex(invalid))
);
}
assertThat(
"expected either none or all valid indices to appear atomically across all projects",
allCreatedInProject,
anyOf(equalTo(0), equalTo(projectsWithValidIndices))
);
}

private void addIndexToAllProjects(MetadataCreateIndexService service, String indexName, Set<ProjectId> projects) {
for (final var projectId : projects) {
service.createIndex(
TimeValue.THIRTY_SECONDS,
TimeValue.THIRTY_SECONDS,
TimeValue.THIRTY_SECONDS,
new CreateIndexClusterStateUpdateRequest("create-pre-existing", projectId, indexName, randomAlphaOfLength(20)).settings(
Settings.builder().put("index.routing.allocation.enable", "none").build()
).waitForActiveShards(ActiveShardCount.NONE),
assertNoFailureListener(r -> {})
);
}
awaitClusterState(
state -> projects.stream()
.allMatch(p -> state.metadata().projects().containsKey(p) && state.metadata().getProject(p).hasIndex(indexName))
);
}

private void addProjectsToClusterState(ClusterService clusterService, Set<ProjectId> projects) {
clusterService.submitUnbatchedStateUpdateTask("init-projects", new ClusterStateUpdateTask() {
@Override
public ClusterState execute(ClusterState currentState) {
final var metadataBuilder = Metadata.builder(currentState.metadata());
projects.forEach(p -> metadataBuilder.put(ProjectMetadata.builder(p)));
return ClusterState.builder(currentState)
.metadata(metadataBuilder)
.routingTable(currentState.globalRoutingTable().initializeProjects(projects))
.build();
}

@Override
public void onFailure(Exception e) {
throw new AssertionError("failed to initialize projects", e);
}
});
awaitClusterState(state -> state.metadata().projects().keySet().containsAll(projects));
}

private ActionListener<ShardsAcknowledgedResponse> listenerForCreateIndexResult(CreateIndexResult expectedResult) {
return switch (expectedResult) {
case SUCCESS -> assertNoFailureListener(response -> assertTrue(response.isAcknowledged()));
case INVALID_NAME -> assertNoSuccessListener(
e -> assertThat(ExceptionsHelper.unwrapCause(e), instanceOf(InvalidIndexNameException.class))
);
case INVALID_SETTINGS -> assertNoSuccessListener(
e -> assertThat(ExceptionsHelper.unwrapCause(e), instanceOf(IllegalArgumentException.class))
);
case ALREADY_EXISTS -> assertNoSuccessListener(
e -> assertThat(ExceptionsHelper.unwrapCause(e), instanceOf(ResourceAlreadyExistsException.class))
);
};
}

private String addRandomIndexNameNoCollision(Set<String> existingIndexNames) {
var indexName = randomIndexName();
while (existingIndexNames.add(indexName) == false) {
Expand Down
Loading