Skip to content

Commit 74654a2

Browse files
brwebrwe
committed
Fix mapping creation on bulk request
When a bulk request triggers an index creation a mapping might not be created. The reason is that when indexing documents in a bulk, an indexing operation might fail due to a shard not yet being started. The mapping service, however, might already have the mapping but the mapping update is never issued to the master, even on subsequent indexing of documents. Instead, the mapping must be propagated to master even if the indexing fails due to a shard not being started. closes #5623
1 parent d828c21 commit 74654a2

File tree

3 files changed

+161
-44
lines changed

3 files changed

+161
-44
lines changed

‎src/main/java/org/elasticsearch/action/bulk/TransportShardBulkAction.java

Lines changed: 67 additions & 44 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import com.google.common.collect.Sets;
2323
import org.elasticsearch.ElasticsearchException;
2424
import org.elasticsearch.ElasticsearchIllegalStateException;
25+
import org.elasticsearch.ElasticsearchWrapperException;
2526
import org.elasticsearch.ExceptionsHelper;
2627
import org.elasticsearch.action.ActionListener;
2728
import org.elasticsearch.action.ActionRequest;
@@ -44,6 +45,7 @@
4445
import org.elasticsearch.cluster.metadata.MappingMetaData;
4546
import org.elasticsearch.cluster.node.DiscoveryNode;
4647
import org.elasticsearch.cluster.routing.ShardIterator;
48+
import org.elasticsearch.common.Nullable;
4749
import org.elasticsearch.common.bytes.BytesReference;
4850
import org.elasticsearch.common.collect.Tuple;
4951
import org.elasticsearch.common.inject.Inject;
@@ -141,7 +143,7 @@ protected PrimaryResponse<BulkShardResponse, BulkShardRequest> shardOperationOnP
141143
final BulkShardRequest request = shardRequest.request;
142144
IndexShard indexShard = indicesService.indexServiceSafe(shardRequest.request.index()).shardSafe(shardRequest.shardId);
143145
Engine.IndexingOperation[] ops = null;
144-
Set<Tuple<String, String>> mappingsToUpdate = null;
146+
final Set<Tuple<String, String>> mappingsToUpdate = Sets.newHashSet();
145147

146148
BulkItemResponse[] responses = new BulkItemResponse[request.items().length];
147149
long[] preVersions = new long[request.items().length];
@@ -150,22 +152,28 @@ protected PrimaryResponse<BulkShardResponse, BulkShardRequest> shardOperationOnP
150152
if (item.request() instanceof IndexRequest) {
151153
IndexRequest indexRequest = (IndexRequest) item.request();
152154
try {
153-
WriteResult result = shardIndexOperation(request, indexRequest, clusterState, indexShard, true);
154-
// add the response
155-
IndexResponse indexResponse = result.response();
156-
responses[requestIndex] = new BulkItemResponse(item.id(), indexRequest.opType().lowercase(), indexResponse);
157-
preVersions[requestIndex] = result.preVersion;
158-
if (result.mappingToUpdate != null) {
159-
if (mappingsToUpdate == null) {
160-
mappingsToUpdate = Sets.newHashSet();
155+
156+
try {
157+
WriteResult result = shardIndexOperation(request, indexRequest, clusterState, indexShard, true);
158+
// add the response
159+
IndexResponse indexResponse = result.response();
160+
responses[requestIndex] = new BulkItemResponse(item.id(), indexRequest.opType().lowercase(), indexResponse);
161+
preVersions[requestIndex] = result.preVersion;
162+
if (result.mappingToUpdate != null) {
163+
mappingsToUpdate.add(result.mappingToUpdate);
161164
}
162-
mappingsToUpdate.add(result.mappingToUpdate);
163-
}
164-
if (result.op != null) {
165-
if (ops == null) {
166-
ops = new Engine.IndexingOperation[request.items().length];
165+
if (result.op != null) {
166+
if (ops == null) {
167+
ops = new Engine.IndexingOperation[request.items().length];
168+
}
169+
ops[requestIndex] = result.op;
170+
}
171+
} catch (WriteFailure e){
172+
Tuple<String, String> mappingsToUpdateOnFailure = e.mappingsToUpdate;
173+
if (mappingsToUpdateOnFailure != null) {
174+
mappingsToUpdate.add(mappingsToUpdateOnFailure);
167175
}
168-
ops[requestIndex] = result.op;
176+
throw e.getCause();
169177
}
170178
} catch (Throwable e) {
171179
// rethrow the failure if we are going to retry on primary and let parent failure to handle it
@@ -174,6 +182,9 @@ protected PrimaryResponse<BulkShardResponse, BulkShardRequest> shardOperationOnP
174182
for (int j = 0; j < requestIndex; j++) {
175183
applyVersion(request.items()[j], preVersions[j]);
176184
}
185+
for (Tuple<String, String> mappingToUpdate : mappingsToUpdate) {
186+
updateMappingOnMaster(mappingToUpdate.v1(), mappingToUpdate.v2());
187+
}
177188
throw (ElasticsearchException) e;
178189
}
179190
if (e instanceof ElasticsearchException && ((ElasticsearchException) e).status() == RestStatus.CONFLICT) {
@@ -239,9 +250,6 @@ protected PrimaryResponse<BulkShardResponse, BulkShardRequest> shardOperationOnP
239250
responses[requestIndex] = new BulkItemResponse(item.id(), "update", updateResponse);
240251
preVersions[requestIndex] = result.preVersion;
241252
if (result.mappingToUpdate != null) {
242-
if (mappingsToUpdate == null) {
243-
mappingsToUpdate = Sets.newHashSet();
244-
}
245253
mappingsToUpdate.add(result.mappingToUpdate);
246254
}
247255
if (result.op != null) {
@@ -277,8 +285,6 @@ protected PrimaryResponse<BulkShardResponse, BulkShardRequest> shardOperationOnP
277285
// we can't try any more
278286
responses[requestIndex] = new BulkItemResponse(item.id(), "update",
279287
new BulkItemResponse.Failure(updateRequest.index(), updateRequest.type(), updateRequest.id(), t));
280-
;
281-
282288
request.items()[requestIndex] = null; // do not send to replicas
283289
}
284290
} else {
@@ -331,10 +337,8 @@ protected PrimaryResponse<BulkShardResponse, BulkShardRequest> shardOperationOnP
331337

332338
}
333339

334-
if (mappingsToUpdate != null) {
335-
for (Tuple<String, String> mappingToUpdate : mappingsToUpdate) {
336-
updateMappingOnMaster(mappingToUpdate.v1(), mappingToUpdate.v2());
337-
}
340+
for (Tuple<String, String> mappingToUpdate : mappingsToUpdate) {
341+
updateMappingOnMaster(mappingToUpdate.v1(), mappingToUpdate.v2());
338342
}
339343

340344
if (request.refresh()) {
@@ -369,6 +373,17 @@ <T> T response() {
369373

370374
}
371375

376+
static class WriteFailure extends ElasticsearchException implements ElasticsearchWrapperException {
377+
@Nullable
378+
final Tuple<String, String> mappingsToUpdate;
379+
380+
WriteFailure(Throwable cause, Tuple<String, String> mappingsToUpdate) {
381+
super(null, cause);
382+
assert cause != null;
383+
this.mappingsToUpdate = mappingsToUpdate;
384+
}
385+
}
386+
372387
private WriteResult shardIndexOperation(BulkShardRequest request, IndexRequest indexRequest, ClusterState clusterState,
373388
IndexShard indexShard, boolean processed) {
374389

@@ -387,30 +402,38 @@ private WriteResult shardIndexOperation(BulkShardRequest request, IndexRequest i
387402
SourceToParse sourceToParse = SourceToParse.source(SourceToParse.Origin.PRIMARY, indexRequest.source()).type(indexRequest.type()).id(indexRequest.id())
388403
.routing(indexRequest.routing()).parent(indexRequest.parent()).timestamp(indexRequest.timestamp()).ttl(indexRequest.ttl());
389404

405+
// update mapping on master if needed, we won't update changes to the same type, since once its changed, it won't have mappers added
406+
Tuple<String, String> mappingsToUpdate = null;
407+
390408
long version;
391409
boolean created;
392410
Engine.IndexingOperation op;
393-
if (indexRequest.opType() == IndexRequest.OpType.INDEX) {
394-
Engine.Index index = indexShard.prepareIndex(sourceToParse).version(indexRequest.version()).versionType(indexRequest.versionType()).origin(Engine.Operation.Origin.PRIMARY);
395-
indexShard.index(index);
396-
version = index.version();
397-
op = index;
398-
created = index.created();
399-
} else {
400-
Engine.Create create = indexShard.prepareCreate(sourceToParse).version(indexRequest.version()).versionType(indexRequest.versionType()).origin(Engine.Operation.Origin.PRIMARY);
401-
indexShard.create(create);
402-
version = create.version();
403-
op = create;
404-
created = true;
405-
}
406411
long preVersion = indexRequest.version();
407-
// update the version on request so it will happen on the replicas
408-
indexRequest.version(version);
409-
410-
// update mapping on master if needed, we won't update changes to the same type, since once its changed, it won't have mappers added
411-
Tuple<String, String> mappingsToUpdate = null;
412-
if (op.parsedDoc().mappingsModified()) {
413-
mappingsToUpdate = Tuple.tuple(indexRequest.index(), indexRequest.type());
412+
try {
413+
if (indexRequest.opType() == IndexRequest.OpType.INDEX) {
414+
Engine.Index index = indexShard.prepareIndex(sourceToParse).version(indexRequest.version()).versionType(indexRequest.versionType()).origin(Engine.Operation.Origin.PRIMARY);
415+
if (index.parsedDoc().mappingsModified()) {
416+
mappingsToUpdate = Tuple.tuple(indexRequest.index(), indexRequest.type());
417+
}
418+
indexShard.index(index);
419+
version = index.version();
420+
op = index;
421+
created = index.created();
422+
} else {
423+
Engine.Create create = indexShard.prepareCreate(sourceToParse).version(indexRequest.version()).versionType(indexRequest.versionType()).origin(Engine.Operation.Origin.PRIMARY);
424+
if (create.parsedDoc().mappingsModified()) {
425+
mappingsToUpdate = Tuple.tuple(indexRequest.index(), indexRequest.type());
426+
}
427+
indexShard.create(create);
428+
version = create.version();
429+
op = create;
430+
created = true;
431+
}
432+
// update the version on request so it will happen on the replicas
433+
indexRequest.versionType(indexRequest.versionType());
434+
indexRequest.version(version);
435+
} catch (Throwable t) {
436+
throw new WriteFailure(t, mappingsToUpdate);
414437
}
415438

416439
IndexResponse indexResponse = new IndexResponse(indexRequest.index(), indexRequest.type(), indexRequest.id(), version, created);

‎src/test/java/org/elasticsearch/action/bulk/BulkIntegrationTests.java

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,70 @@
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+
20+
21+
package org.elasticsearch.action.bulk;
22+
23+
import com.google.common.base.Charsets;
24+
import com.google.common.base.Predicate;
25+
import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsRequest;
26+
import org.elasticsearch.action.admin.indices.mapping.get.GetMappingsResponse;
27+
import org.elasticsearch.test.ElasticsearchIntegrationTest;
28+
import static org.hamcrest.Matchers.*;
29+
import static org.elasticsearch.common.io.Streams.copyToStringFromClasspath;
30+
31+
import org.junit.Test;
32+
33+
@ElasticsearchIntegrationTest.ClusterScope(scope= ElasticsearchIntegrationTest.Scope.SUITE, numNodes=1)
34+
public class BulkIntegrationTests extends ElasticsearchIntegrationTest{
35+
36+
@Test
37+
public void testBulkIndexCreatesMapping() throws Exception {
38+
String bulkAction = copyToStringFromClasspath("/org/elasticsearch/action/bulk/bulk-log.json");
39+
BulkRequestBuilder bulkBuilder = new BulkRequestBuilder(client());
40+
bulkBuilder.add(bulkAction.getBytes(Charsets.UTF_8), 0, bulkAction.length(), true, null, null);
41+
bulkBuilder.execute().actionGet();
42+
awaitBusy(new Predicate<Object>() {
43+
@Override
44+
public boolean apply(Object input) {
45+
try {
46+
GetMappingsResponse mappingsResponse = client().admin().indices().getMappings(new GetMappingsRequest()).get();
47+
return mappingsResponse.getMappings().containsKey("logstash-2014.03.30");
48+
} catch (Throwable t) {
49+
return false;
50+
}
51+
}
52+
});
53+
awaitBusy(new Predicate<Object>() {
54+
@Override
55+
public boolean apply(Object input) {
56+
try {
57+
GetMappingsResponse mappingsResponse = client().admin().indices().getMappings(new GetMappingsRequest()).get();
58+
return mappingsResponse.getMappings().get("logstash-2014.03.30").containsKey("logs");
59+
} catch (Throwable t) {
60+
return false;
61+
}
62+
}
63+
});
64+
ensureYellow();
65+
GetMappingsResponse mappingsResponse = client().admin().indices().getMappings(new GetMappingsRequest()).get();
66+
assertThat(mappingsResponse.mappings().size(), equalTo(1));
67+
assertTrue(mappingsResponse.getMappings().containsKey("logstash-2014.03.30"));
68+
assertTrue(mappingsResponse.getMappings().get("logstash-2014.03.30").containsKey("logs"));
69+
}
70+
}

‎src/test/java/org/elasticsearch/action/bulk/bulk-log.json

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,24 @@
1+
{"index":{"_index":"logstash-2014.03.30","_type":"logs"}}
2+
{"message":"in24.inetnebr.com--[01/Aug/1995:00:00:01-0400]\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"2001839","@version":"1","@timestamp":"2014-03-30T12:38:10.048Z","host":["romeo","in24.inetnebr.com"],"monthday":1,"month":8,"year":1995,"time":"00:00:01","tz":"-0400","request":"\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"","httpresponse":"200","size":1839,"rtime":"1995-08-01T00:00:01.000Z"}
3+
{"index":{"_index":"logstash-2014.03.30","_type":"logs"}}
4+
{"message":"in24.inetnebr.com--[01/Aug/1995:00:00:01-0400]\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"2001839","@version":"1","@timestamp":"2014-03-30T12:38:10.048Z","host":["romeo","in24.inetnebr.com"],"monthday":1,"month":8,"year":1995,"time":"00:00:01","tz":"-0400","request":"\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"","httpresponse":"200","size":1839,"rtime":"1995-08-01T00:00:01.000Z"}
5+
{"index":{"_index":"logstash-2014.03.30","_type":"logs"}}
6+
{"message":"in24.inetnebr.com--[01/Aug2/1995:00:00:01-0400]\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"2001839","@version":"1","@timestamp":"2014-03-30T12:38:10.048Z","host":["romeo","in24.inetnebr.com"],"monthday":1,"month":8,"year":1995,"time":"00:00:01","tz":"-0400","request":"\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"","httpresponse":"200","size":1839,"rtime":"1995-08-01T00:00:01.000Z"}
7+
{"index":{"_index":"logstash-2014.03.30","_type":"logs"}}
8+
{"message":"in24.inetnebr.com--[01/Aug/1995:00:00:01-0400]\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"2001839","@version":"1","@timestamp":"2014-03-30T12:38:10.048Z","host":["romeo","in24.inetnebr.com"],"monthday":1,"month":8,"year":1995,"time":"00:00:01","tz":"-0400","request":"\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"","httpresponse":"200","size":1839,"rtime":"1995-08-01T00:00:01.000Z"}
9+
{"index":{"_index":"logstash-2014.03.30","_type":"logs"}}
10+
{"message":"in24.inetnebr.com--[01/Aug/1995:00:00:01-0400]\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"2001839","@version":"1","@timestamp":"2014-03-30T12:38:10.048Z","host":["romeo","in24.inetnebr.com"],"monthday":1,"month":8,"year":1995,"time":"00:00:01","tz":"-0400","request":"\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"","httpresponse":"200","size":1839,"rtime":"1995-08-01T00:00:01.000Z"}
11+
{"index":{"_index":"logstash-2014.03.30","_type":"logs"}}
12+
{"message":"in24.inetnebr.com--[01/Aug2/1995:00:00:01-0400]\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"2001839","@version":"1","@timestamp":"2014-03-30T12:38:10.048Z","host":["romeo","in24.inetnebr.com"],"monthday":1,"month":8,"year":1995,"time":"00:00:01","tz":"-0400","request":"\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"","httpresponse":"200","size":1839,"rtime":"1995-08-01T00:00:01.000Z"}
13+
{"index":{"_index":"logstash-2014.03.30","_type":"logs"}}
14+
{"message":"in24.inetnebr.com--[01/Aug/1995:00:00:01-0400]\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"2001839","@version":"1","@timestamp":"2014-03-30T12:38:10.048Z","host":["romeo","in24.inetnebr.com"],"monthday":1,"month":8,"year":1995,"time":"00:00:01","tz":"-0400","request":"\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"","httpresponse":"200","size":1839,"rtime":"1995-08-01T00:00:01.000Z"}
15+
{"index":{"_index":"logstash-2014.03.30","_type":"logs"}}
16+
{"message":"in24.inetnebr.com--[01/Aug/1995:00:00:01-0400]\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"2001839","@version":"1","@timestamp":"2014-03-30T12:38:10.048Z","host":["romeo","in24.inetnebr.com"],"monthday":1,"month":8,"year":1995,"time":"00:00:01","tz":"-0400","request":"\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"","httpresponse":"200","size":1839,"rtime":"1995-08-01T00:00:01.000Z"}
17+
{"index":{"_index":"logstash-2014.03.30","_type":"logs"}}
18+
{"message":"in24.inetnebr.com--[01/Aug2/1995:00:00:01-0400]\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"2001839","@version":"1","@timestamp":"2014-03-30T12:38:10.048Z","host":["romeo","in24.inetnebr.com"],"monthday":1,"month":8,"year":1995,"time":"00:00:01","tz":"-0400","request":"\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"","httpresponse":"200","size":1839,"rtime":"1995-08-01T00:00:01.000Z"}
19+
{"index":{"_index":"logstash-2014.03.30","_type":"logs"}}
20+
{"message":"in24.inetnebr.com--[01/Aug/1995:00:00:01-0400]\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"2001839","@version":"1","@timestamp":"2014-03-30T12:38:10.048Z","host":["romeo","in24.inetnebr.com"],"monthday":1,"month":8,"year":1995,"time":"00:00:01","tz":"-0400","request":"\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"","httpresponse":"200","size":1839,"rtime":"1995-08-01T00:00:01.000Z"}
21+
{"index":{"_index":"logstash-2014.03.30","_type":"logs"}}
22+
{"message":"in24.inetnebr.com--[01/Aug/1995:00:00:01-0400]\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"2001839","@version":"1","@timestamp":"2014-03-30T12:38:10.048Z","host":["romeo","in24.inetnebr.com"],"monthday":1,"month":8,"year":1995,"time":"00:00:01","tz":"-0400","request":"\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"","httpresponse":"200","size":1839,"rtime":"1995-08-01T00:00:01.000Z"}
23+
{"index":{"_index":"logstash-2014.03.30","_type":"logs"}}
24+
{"message":"in24.inetnebr.com--[01/Aug2/1995:00:00:01-0400]\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"2001839","@version":"1","@timestamp":"2014-03-30T12:38:10.048Z","host":["romeo","in24.inetnebr.com"],"monthday":1,"month":8,"year":1995,"time":"00:00:01","tz":"-0400","request":"\"GET/shuttle/missions/sts-68/news/sts-68-mcc-05.txtHTTP/1.0\"","httpresponse":"200","size":1839,"rtime":"1995-08-01T00:00:01.000Z"}

0 commit comments

Comments
 (0)