Skip to content

Commit 7b98184

Browse files
ywelschywelsch
committed
Block too many concurrent mapping updates (#51038)
Ensures that there are not too many concurrent dynamic mapping updates going out from the data nodes to the master. Closes #50670
1 parent ac26392 commit 7b98184

File tree

5 files changed

+190
-1
lines changed

5 files changed

+190
-1
lines changed

‎server/src/main/java/org/elasticsearch/cluster/action/index/MappingUpdatedAction.java‎

Lines changed: 68 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,11 +32,14 @@
3232
import org.elasticsearch.common.settings.Settings;
3333
import org.elasticsearch.common.unit.TimeValue;
3434
import org.elasticsearch.common.util.concurrent.UncategorizedExecutionException;
35+
import org.elasticsearch.common.util.concurrent.RunOnce;
3536
import org.elasticsearch.common.xcontent.XContentType;
3637
import org.elasticsearch.index.Index;
3738
import org.elasticsearch.index.mapper.MapperService;
3839
import org.elasticsearch.index.mapper.Mapping;
3940

41+
import java.util.concurrent.Semaphore;
42+
4043
/**
4144
* Called by shards in the cluster when their mapping was dynamically updated and it needs to be updated
4245
* in the cluster state meta data (and broadcast to all members).
@@ -47,19 +50,30 @@ public class MappingUpdatedAction {
4750
Setting.positiveTimeSetting("indices.mapping.dynamic_timeout", TimeValue.timeValueSeconds(30),
4851
Property.Dynamic, Property.NodeScope);
4952

53+
public static final Setting<Integer> INDICES_MAX_IN_FLIGHT_UPDATES_SETTING =
54+
Setting.intSetting("indices.mapping.max_in_flight_updates", 10, 1, 1000,
55+
Property.Dynamic, Property.NodeScope);
56+
5057
private IndicesAdminClient client;
5158
private volatile TimeValue dynamicMappingUpdateTimeout;
59+
private final AdjustableSemaphore semaphore;
5260

5361
@Inject
5462
public MappingUpdatedAction(Settings settings, ClusterSettings clusterSettings) {
5563
this.dynamicMappingUpdateTimeout = INDICES_MAPPING_DYNAMIC_TIMEOUT_SETTING.get(settings);
64+
this.semaphore = new AdjustableSemaphore(INDICES_MAX_IN_FLIGHT_UPDATES_SETTING.get(settings), true);
5665
clusterSettings.addSettingsUpdateConsumer(INDICES_MAPPING_DYNAMIC_TIMEOUT_SETTING, this::setDynamicMappingUpdateTimeout);
66+
clusterSettings.addSettingsUpdateConsumer(INDICES_MAX_IN_FLIGHT_UPDATES_SETTING, this::setMaxInFlightUpdates);
5767
}
5868

5969
private void setDynamicMappingUpdateTimeout(TimeValue dynamicMappingUpdateTimeout) {
6070
this.dynamicMappingUpdateTimeout = dynamicMappingUpdateTimeout;
6171
}
6272

73+
private void setMaxInFlightUpdates(int maxInFlightUpdates) {
74+
semaphore.setMaxPermits(maxInFlightUpdates);
75+
}
76+
6377
public void setClient(Client client) {
6478
this.client = client.admin().indices();
6579
}
@@ -74,7 +88,34 @@ public void updateMappingOnMaster(Index index, String type, Mapping mappingUpdat
7488
if (type.equals(MapperService.DEFAULT_MAPPING)) {
7589
throw new IllegalArgumentException("_default_ mapping should not be updated");
7690
}
77-
client.preparePutMapping().setConcreteIndex(index).setType(type).setSource(mappingUpdate.toString(), XContentType.JSON)
91+
92+
final RunOnce release = new RunOnce(() -> semaphore.release());
93+
try {
94+
semaphore.acquire();
95+
} catch (InterruptedException e) {
96+
Thread.currentThread().interrupt();
97+
listener.onFailure(e);
98+
return;
99+
}
100+
boolean successFullySent = false;
101+
try {
102+
sendUpdateMapping(index, type, mappingUpdate, ActionListener.runBefore(listener, release::run));
103+
successFullySent = true;
104+
} finally {
105+
if (successFullySent == false) {
106+
release.run();
107+
}
108+
}
109+
}
110+
111+
// used by tests
112+
int blockedThreads() {
113+
return semaphore.getQueueLength();
114+
}
115+
116+
// can be overridden by tests
117+
protected void sendUpdateMapping(Index index, String type, Mapping mappingUpdate, ActionListener<Void> listener) {
118+
client.preparePutMapping().setConcreteIndex(index).setType(type).setSource(mappingUpdate.toString(), XContentType.JSON)
78119
.setMasterNodeTimeout(dynamicMappingUpdateTimeout).setTimeout(TimeValue.ZERO)
79120
.execute(new ActionListener<AcknowledgedResponse>() {
80121
@Override
@@ -101,4 +142,30 @@ private static RuntimeException unwrapEsException(ElasticsearchException esEx) {
101142
}
102143
return new UncategorizedExecutionException("Failed execution", root);
103144
}
145+
146+
static class AdjustableSemaphore extends Semaphore {
147+
148+
private final Object maxPermitsMutex = new Object();
149+
private int maxPermits;
150+
151+
AdjustableSemaphore(int maxPermits, boolean fair) {
152+
super(maxPermits, fair);
153+
this.maxPermits = maxPermits;
154+
}
155+
156+
void setMaxPermits(int permits) {
157+
synchronized (maxPermitsMutex) {
158+
final int diff = Math.subtractExact(permits, maxPermits);
159+
if (diff > 0) {
160+
// add permits
161+
release(diff);
162+
} else if (diff < 0) {
163+
// remove permits
164+
reducePermits(Math.negateExact(diff));
165+
}
166+
167+
maxPermits = permits;
168+
}
169+
}
170+
}
104171
}

‎server/src/main/java/org/elasticsearch/common/settings/ClusterSettings.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -209,6 +209,7 @@ public void apply(Settings value, Settings current, Settings previous) {
209209
IndicesQueryCache.INDICES_CACHE_QUERY_COUNT_SETTING,
210210
IndicesQueryCache.INDICES_QUERIES_CACHE_ALL_SEGMENTS_SETTING,
211211
MappingUpdatedAction.INDICES_MAPPING_DYNAMIC_TIMEOUT_SETTING,
212+
MappingUpdatedAction.INDICES_MAX_IN_FLIGHT_UPDATES_SETTING,
212213
MetaData.SETTING_READ_ONLY_SETTING,
213214
MetaData.SETTING_READ_ONLY_ALLOW_DELETE_SETTING,
214215
MetaData.SETTING_CLUSTER_MAX_SHARDS_PER_NODE,

‎server/src/test/java/org/elasticsearch/cluster/action/index/MappingUpdatedActionTests.java‎

Lines changed: 118 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,118 @@
1+
/*
2+
* Licensed to Elasticsearch under one or more contributor
3+
* license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright
5+
* ownership. Elasticsearch licenses this file to you under
6+
* the Apache License, Version 2.0 (the "License"); you may
7+
* not use this file except in compliance with the License.
8+
* You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
package org.elasticsearch.cluster.action.index;
20+
21+
import org.elasticsearch.action.ActionListener;
22+
import org.elasticsearch.action.support.PlainActionFuture;
23+
import org.elasticsearch.cluster.action.index.MappingUpdatedAction.AdjustableSemaphore;
24+
import org.elasticsearch.common.settings.ClusterSettings;
25+
import org.elasticsearch.common.settings.Settings;
26+
import org.elasticsearch.index.Index;
27+
import org.elasticsearch.index.mapper.Mapping;
28+
import org.elasticsearch.test.ESTestCase;
29+
30+
import java.util.List;
31+
import java.util.concurrent.CopyOnWriteArrayList;
32+
33+
public class MappingUpdatedActionTests extends ESTestCase {
34+
35+
public void testAdjustableSemaphore() {
36+
AdjustableSemaphore sem = new AdjustableSemaphore(1, randomBoolean());
37+
assertEquals(1, sem.availablePermits());
38+
assertTrue(sem.tryAcquire());
39+
assertEquals(0, sem.availablePermits());
40+
assertFalse(sem.tryAcquire());
41+
assertEquals(0, sem.availablePermits());
42+
43+
// increase the number of max permits to 2
44+
sem.setMaxPermits(2);
45+
assertEquals(1, sem.availablePermits());
46+
assertTrue(sem.tryAcquire());
47+
assertEquals(0, sem.availablePermits());
48+
49+
// release all current permits
50+
sem.release();
51+
assertEquals(1, sem.availablePermits());
52+
sem.release();
53+
assertEquals(2, sem.availablePermits());
54+
55+
// reduce number of max permits to 1
56+
sem.setMaxPermits(1);
57+
assertEquals(1, sem.availablePermits());
58+
// set back to 2
59+
sem.setMaxPermits(2);
60+
assertEquals(2, sem.availablePermits());
61+
62+
// take both permits and reduce max permits
63+
assertTrue(sem.tryAcquire());
64+
assertTrue(sem.tryAcquire());
65+
assertEquals(0, sem.availablePermits());
66+
assertFalse(sem.tryAcquire());
67+
sem.setMaxPermits(1);
68+
assertEquals(-1, sem.availablePermits());
69+
assertFalse(sem.tryAcquire());
70+
71+
// release one permit
72+
sem.release();
73+
assertEquals(0, sem.availablePermits());
74+
assertFalse(sem.tryAcquire());
75+
76+
// release second permit
77+
sem.release();
78+
assertEquals(1, sem.availablePermits());
79+
assertTrue(sem.tryAcquire());
80+
}
81+
82+
public void testMappingUpdatedActionBlocks() throws Exception {
83+
List<ActionListener<Void>> inFlightListeners = new CopyOnWriteArrayList<>();
84+
final MappingUpdatedAction mua = new MappingUpdatedAction(Settings.builder()
85+
.put(MappingUpdatedAction.INDICES_MAX_IN_FLIGHT_UPDATES_SETTING.getKey(), 1).build(),
86+
new ClusterSettings(Settings.EMPTY, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS)) {
87+
88+
@Override
89+
protected void sendUpdateMapping(Index index, String type, Mapping mappingUpdate, ActionListener<Void> listener) {
90+
inFlightListeners.add(listener);
91+
}
92+
};
93+
94+
PlainActionFuture<Void> fut1 = new PlainActionFuture<>();
95+
mua.updateMappingOnMaster(null, "test", null, fut1);
96+
assertEquals(1, inFlightListeners.size());
97+
assertEquals(0, mua.blockedThreads());
98+
99+
PlainActionFuture<Void> fut2 = new PlainActionFuture<>();
100+
Thread thread = new Thread(() -> {
101+
mua.updateMappingOnMaster(null, "test", null, fut2); // blocked
102+
});
103+
thread.start();
104+
assertBusy(() -> assertEquals(1, mua.blockedThreads()));
105+
106+
assertEquals(1, inFlightListeners.size());
107+
assertFalse(fut1.isDone());
108+
inFlightListeners.remove(0).onResponse(null);
109+
assertTrue(fut1.isDone());
110+
111+
thread.join();
112+
assertEquals(0, mua.blockedThreads());
113+
assertEquals(1, inFlightListeners.size());
114+
assertFalse(fut2.isDone());
115+
inFlightListeners.remove(0).onResponse(null);
116+
assertTrue(fut2.isDone());
117+
}
118+
}

‎server/src/test/java/org/elasticsearch/snapshots/SnapshotResiliencyTests.java‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -774,6 +774,7 @@ private Environment createEnvironment(String nodeName) {
774774
.put(Environment.PATH_REPO_SETTING.getKey(), tempDir.resolve("repo").toAbsolutePath())
775775
.putList(ClusterBootstrapService.INITIAL_MASTER_NODES_SETTING.getKey(),
776776
ClusterBootstrapService.INITIAL_MASTER_NODES_SETTING.get(Settings.EMPTY))
777+
.put(MappingUpdatedAction.INDICES_MAX_IN_FLIGHT_UPDATES_SETTING.getKey(), 1000) // o.w. some tests might block
777778
.build());
778779
}
779780

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -489,6 +489,8 @@ private static Settings getRandomNodeSettings(long seed) {
489489
if (random.nextBoolean()) {
490490
builder.put(MappingUpdatedAction.INDICES_MAPPING_DYNAMIC_TIMEOUT_SETTING.getKey(),
491491
timeValueSeconds(RandomNumbers.randomIntBetween(random, 10, 30)).getStringRep());
492+
builder.put(MappingUpdatedAction.INDICES_MAX_IN_FLIGHT_UPDATES_SETTING.getKey(),
493+
RandomNumbers.randomIntBetween(random, 1, 10));
492494
}
493495

494496
// turning on the real memory circuit breaker leads to spurious test failures. As have no full control over heap usage, we

0 commit comments

Comments
 (0)