Skip to content

Commit 4344305

Browse files
markharwoodmarkharwood
authored
Type removal - added deprecation warnings to _bulk apis (#36549)
Added warnings checks to existing tests Added “defaultTypeIfNull” to DocWriteRequest interface so that Bulk requests can override a null choice of document type with any global custom choice. Related to #35190
1 parent 04dcb13 commit 4344305

File tree

46 files changed

+498
-164
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

46 files changed

+498
-164
lines changed

‎client/client-benchmark-noop-api-plugin/src/main/java/org/elasticsearch/plugin/noop/action/bulk/TransportNoopBulkAction.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -32,13 +32,15 @@
3232
import org.elasticsearch.tasks.Task;
3333
import org.elasticsearch.transport.TransportService;
3434

35+
import java.util.function.Supplier;
36+
3537
public class TransportNoopBulkAction extends HandledTransportAction<BulkRequest, BulkResponse> {
3638
private static final BulkItemResponse ITEM_RESPONSE = new BulkItemResponse(1, DocWriteRequest.OpType.UPDATE,
3739
new UpdateResponse(new ShardId("mock", "", 1), "mock_type", "1", 1L, DocWriteResponse.Result.CREATED));
3840

3941
@Inject
4042
public TransportNoopBulkAction(TransportService transportService, ActionFilters actionFilters) {
41-
super(NoopBulkAction.NAME, transportService, actionFilters, BulkRequest::new);
43+
super(NoopBulkAction.NAME, transportService, actionFilters, (Supplier<BulkRequest>) BulkRequest::new);
4244
}
4345

4446
@Override

‎client/rest-high-level/src/main/java/org/elasticsearch/client/RequestConverters.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -165,7 +165,9 @@ static Request bulk(BulkRequest bulkRequest) throws IOException {
165165
metadata.field("_index", action.index());
166166
}
167167
if (Strings.hasLength(action.type())) {
168-
metadata.field("_type", action.type());
168+
if (MapperService.SINGLE_MAPPING_NAME.equals(action.type()) == false) {
169+
metadata.field("_type", action.type());
170+
}
169171
}
170172
if (Strings.hasLength(action.id())) {
171173
metadata.field("_id", action.id());

‎client/rest-high-level/src/test/java/org/elasticsearch/client/BulkProcessorIT.java‎

Lines changed: 114 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,8 @@
3535
import org.elasticsearch.common.unit.ByteSizeValue;
3636
import org.elasticsearch.common.unit.TimeValue;
3737
import org.elasticsearch.common.xcontent.XContentType;
38+
import org.elasticsearch.index.mapper.MapperService;
39+
import org.elasticsearch.rest.action.document.RestBulkAction;
3840
import org.elasticsearch.search.SearchHit;
3941
import org.hamcrest.Matcher;
4042
import org.hamcrest.Matchers;
@@ -70,8 +72,15 @@ public class BulkProcessorIT extends ESRestHighLevelClientTestCase {
7072

7173
private static BulkProcessor.Builder initBulkProcessorBuilder(BulkProcessor.Listener listener) {
7274
return BulkProcessor.builder(
73-
(request, bulkListener) -> highLevelClient().bulkAsync(request, RequestOptions.DEFAULT, bulkListener), listener);
75+
(request, bulkListener) -> highLevelClient().bulkAsync(request, RequestOptions.DEFAULT,
76+
bulkListener), listener);
7477
}
78+
79+
private static BulkProcessor.Builder initBulkProcessorBuilderUsingTypes(BulkProcessor.Listener listener) {
80+
return BulkProcessor.builder(
81+
(request, bulkListener) -> highLevelClient().bulkAsync(request, expectWarnings(RestBulkAction.TYPES_DEPRECATION_MESSAGE),
82+
bulkListener), listener);
83+
}
7584

7685
public void testThatBulkProcessorCountIsCorrect() throws Exception {
7786
final CountDownLatch latch = new CountDownLatch(1);
@@ -320,35 +329,105 @@ public void testGlobalParametersAndSingleRequest() throws Exception {
320329
public void testGlobalParametersAndBulkProcessor() throws Exception {
321330
createIndexWithMultipleShards("test");
322331

323-
final CountDownLatch latch = new CountDownLatch(1);
324-
BulkProcessorTestListener listener = new BulkProcessorTestListener(latch);
325332
createFieldAddingPipleine("pipeline_id", "fieldNameXYZ", "valueXYZ");
333+
final String customType = "testType";
334+
final String ignoredType = "ignoredType";
326335

327336
int numDocs = randomIntBetween(10, 10);
328-
try (BulkProcessor processor = initBulkProcessorBuilder(listener)
329-
//let's make sure that the bulk action limit trips, one single execution will index all the documents
330-
.setConcurrentRequests(randomIntBetween(0, 1)).setBulkActions(numDocs)
331-
.setFlushInterval(TimeValue.timeValueHours(24)).setBulkSize(new ByteSizeValue(1, ByteSizeUnit.GB))
332-
.setGlobalIndex("test")
333-
.setGlobalType("_doc")
334-
.setGlobalRouting("routing")
335-
.setGlobalPipeline("pipeline_id")
336-
.build()) {
337-
338-
indexDocs(processor, numDocs, null, null, "test", "pipeline_id");
339-
latch.await();
340-
341-
assertThat(listener.beforeCounts.get(), equalTo(1));
342-
assertThat(listener.afterCounts.get(), equalTo(1));
343-
assertThat(listener.bulkFailures.size(), equalTo(0));
344-
assertResponseItems(listener.bulkItems, numDocs);
345-
346-
Iterable<SearchHit> hits = searchAll(new SearchRequest("test").routing("routing"));
337+
{
338+
final CountDownLatch latch = new CountDownLatch(1);
339+
BulkProcessorTestListener listener = new BulkProcessorTestListener(latch);
340+
//Check that untyped document additions inherit the global type
341+
String globalType = customType;
342+
String localType = null;
343+
try (BulkProcessor processor = initBulkProcessorBuilderUsingTypes(listener)
344+
//let's make sure that the bulk action limit trips, one single execution will index all the documents
345+
.setConcurrentRequests(randomIntBetween(0, 1)).setBulkActions(numDocs)
346+
.setFlushInterval(TimeValue.timeValueHours(24)).setBulkSize(new ByteSizeValue(1, ByteSizeUnit.GB))
347+
.setGlobalIndex("test")
348+
.setGlobalType(globalType)
349+
.setGlobalRouting("routing")
350+
.setGlobalPipeline("pipeline_id")
351+
.build()) {
352+
353+
indexDocs(processor, numDocs, null, localType, "test", globalType, "pipeline_id");
354+
latch.await();
355+
356+
assertThat(listener.beforeCounts.get(), equalTo(1));
357+
assertThat(listener.afterCounts.get(), equalTo(1));
358+
assertThat(listener.bulkFailures.size(), equalTo(0));
359+
assertResponseItems(listener.bulkItems, numDocs, globalType);
360+
361+
Iterable<SearchHit> hits = searchAll(new SearchRequest("test").routing("routing"));
362+
363+
assertThat(hits, everyItem(hasProperty(fieldFromSource("fieldNameXYZ"), equalTo("valueXYZ"))));
364+
assertThat(hits, everyItem(Matchers.allOf(hasIndex("test"), hasType(globalType))));
365+
assertThat(hits, containsInAnyOrder(expectedIds(numDocs)));
366+
}
347367

348-
assertThat(hits, everyItem(hasProperty(fieldFromSource("fieldNameXYZ"), equalTo("valueXYZ"))));
349-
assertThat(hits, everyItem(Matchers.allOf(hasIndex("test"), hasType("_doc"))));
350-
assertThat(hits, containsInAnyOrder(expectedIds(numDocs)));
351368
}
369+
{
370+
//Check that typed document additions don't inherit the global type
371+
String globalType = ignoredType;
372+
String localType = customType;
373+
final CountDownLatch latch = new CountDownLatch(1);
374+
BulkProcessorTestListener listener = new BulkProcessorTestListener(latch);
375+
try (BulkProcessor processor = initBulkProcessorBuilderUsingTypes(listener)
376+
//let's make sure that the bulk action limit trips, one single execution will index all the documents
377+
.setConcurrentRequests(randomIntBetween(0, 1)).setBulkActions(numDocs)
378+
.setFlushInterval(TimeValue.timeValueHours(24)).setBulkSize(new ByteSizeValue(1, ByteSizeUnit.GB))
379+
.setGlobalIndex("test")
380+
.setGlobalType(globalType)
381+
.setGlobalRouting("routing")
382+
.setGlobalPipeline("pipeline_id")
383+
.build()) {
384+
indexDocs(processor, numDocs, null, localType, "test", globalType, "pipeline_id");
385+
latch.await();
386+
387+
assertThat(listener.beforeCounts.get(), equalTo(1));
388+
assertThat(listener.afterCounts.get(), equalTo(1));
389+
assertThat(listener.bulkFailures.size(), equalTo(0));
390+
assertResponseItems(listener.bulkItems, numDocs, localType);
391+
392+
Iterable<SearchHit> hits = searchAll(new SearchRequest("test").routing("routing"));
393+
394+
assertThat(hits, everyItem(hasProperty(fieldFromSource("fieldNameXYZ"), equalTo("valueXYZ"))));
395+
assertThat(hits, everyItem(Matchers.allOf(hasIndex("test"), hasType(localType))));
396+
assertThat(hits, containsInAnyOrder(expectedIds(numDocs)));
397+
}
398+
}
399+
{
400+
//Check that untyped document additions and untyped global inherit the established custom type
401+
// (the custom document type introduced to the mapping by the earlier code in this test)
402+
String globalType = null;
403+
String localType = null;
404+
final CountDownLatch latch = new CountDownLatch(1);
405+
BulkProcessorTestListener listener = new BulkProcessorTestListener(latch);
406+
try (BulkProcessor processor = initBulkProcessorBuilder(listener)
407+
//let's make sure that the bulk action limit trips, one single execution will index all the documents
408+
.setConcurrentRequests(randomIntBetween(0, 1)).setBulkActions(numDocs)
409+
.setFlushInterval(TimeValue.timeValueHours(24)).setBulkSize(new ByteSizeValue(1, ByteSizeUnit.GB))
410+
.setGlobalIndex("test")
411+
.setGlobalType(globalType)
412+
.setGlobalRouting("routing")
413+
.setGlobalPipeline("pipeline_id")
414+
.build()) {
415+
indexDocs(processor, numDocs, null, localType, "test", globalType, "pipeline_id");
416+
latch.await();
417+
418+
assertThat(listener.beforeCounts.get(), equalTo(1));
419+
assertThat(listener.afterCounts.get(), equalTo(1));
420+
assertThat(listener.bulkFailures.size(), equalTo(0));
421+
assertResponseItems(listener.bulkItems, numDocs, MapperService.SINGLE_MAPPING_NAME);
422+
423+
Iterable<SearchHit> hits = searchAll(new SearchRequest("test").routing("routing"));
424+
425+
assertThat(hits, everyItem(hasProperty(fieldFromSource("fieldNameXYZ"), equalTo("valueXYZ"))));
426+
assertThat(hits, everyItem(Matchers.allOf(hasIndex("test"), hasType(customType))));
427+
assertThat(hits, containsInAnyOrder(expectedIds(numDocs)));
428+
}
429+
}
430+
assertWarnings(RestBulkAction.TYPES_DEPRECATION_MESSAGE);
352431
}
353432

354433
@SuppressWarnings("unchecked")
@@ -359,15 +438,15 @@ private Matcher<SearchHit>[] expectedIds(int numDocs) {
359438
.<Matcher<SearchHit>>toArray(Matcher[]::new);
360439
}
361440

362-
private static MultiGetRequest indexDocs(BulkProcessor processor, int numDocs, String localIndex,
441+
private static MultiGetRequest indexDocs(BulkProcessor processor, int numDocs, String localIndex, String localType,
363442
String globalIndex, String globalType, String globalPipeline) throws Exception {
364443
MultiGetRequest multiGetRequest = new MultiGetRequest();
365444
for (int i = 1; i <= numDocs; i++) {
366445
if (randomBoolean()) {
367-
processor.add(new IndexRequest(localIndex).id(Integer.toString(i))
446+
processor.add(new IndexRequest(localIndex, localType, Integer.toString(i))
368447
.source(XContentType.JSON, "field", randomRealisticUnicodeOfLengthBetween(1, 30)));
369448
} else {
370-
BytesArray data = bytesBulkRequest(localIndex, "_doc", i);
449+
BytesArray data = bytesBulkRequest(localIndex, localType, i);
371450
processor.add(data, globalIndex, globalType, globalPipeline, null, XContentType.JSON);
372451
}
373452
multiGetRequest.add(localIndex, Integer.toString(i));
@@ -396,15 +475,19 @@ private static BytesArray bytesBulkRequest(String localIndex, String localType,
396475
}
397476

398477
private static MultiGetRequest indexDocs(BulkProcessor processor, int numDocs) throws Exception {
399-
return indexDocs(processor, numDocs, "test", null, null, null);
478+
return indexDocs(processor, numDocs, "test", null, null, null, null);
400479
}
401-
480+
402481
private static void assertResponseItems(List<BulkItemResponse> bulkItemResponses, int numDocs) {
482+
assertResponseItems(bulkItemResponses, numDocs, MapperService.SINGLE_MAPPING_NAME);
483+
}
484+
485+
private static void assertResponseItems(List<BulkItemResponse> bulkItemResponses, int numDocs, String expectedType) {
403486
assertThat(bulkItemResponses.size(), is(numDocs));
404487
int i = 1;
405488
for (BulkItemResponse bulkItemResponse : bulkItemResponses) {
406489
assertThat(bulkItemResponse.getIndex(), equalTo("test"));
407-
assertThat(bulkItemResponse.getType(), equalTo("_doc"));
490+
assertThat(bulkItemResponse.getType(), equalTo(expectedType));
408491
assertThat(bulkItemResponse.getId(), equalTo(Integer.toString(i++)));
409492
assertThat("item " + i + " failed with cause: " + bulkItemResponse.getFailureMessage(),
410493
bulkItemResponse.isFailed(), equalTo(false));

‎client/rest-high-level/src/test/java/org/elasticsearch/client/BulkProcessorRetryIT.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -143,7 +143,7 @@ public void afterBulk(long executionId, BulkRequest request, Throwable failure)
143143
private static MultiGetRequest indexDocs(BulkProcessor processor, int numDocs) {
144144
MultiGetRequest multiGetRequest = new MultiGetRequest();
145145
for (int i = 1; i <= numDocs; i++) {
146-
processor.add(new IndexRequest(INDEX_NAME, "_doc", Integer.toString(i))
146+
processor.add(new IndexRequest(INDEX_NAME).id(Integer.toString(i))
147147
.source(XContentType.JSON, "field", randomRealisticUnicodeOfCodepointLengthBetween(1, 30)));
148148
multiGetRequest.add(INDEX_NAME, Integer.toString(i));
149149
}

‎client/rest-high-level/src/test/java/org/elasticsearch/client/BulkRequestWithGlobalParametersIT.java‎

Lines changed: 29 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.elasticsearch.action.index.IndexRequest;
2525
import org.elasticsearch.action.search.SearchRequest;
2626
import org.elasticsearch.common.xcontent.XContentType;
27+
import org.elasticsearch.rest.action.document.RestBulkAction;
2728
import org.elasticsearch.search.SearchHit;
2829

2930
import java.io.IOException;
@@ -140,28 +141,27 @@ public void testIndexGlobalAndPerRequest() throws IOException {
140141
}
141142

142143
public void testGlobalType() throws IOException {
143-
BulkRequest request = new BulkRequest(null, "_doc");
144+
BulkRequest request = new BulkRequest(null, "global_type");
144145
request.add(new IndexRequest("index").id("1")
145146
.source(XContentType.JSON, "field", "bulk1"));
146147
request.add(new IndexRequest("index").id("2")
147148
.source(XContentType.JSON, "field", "bulk2"));
148149

149-
bulk(request);
150+
bulkWithTypes(request);
150151

151152
Iterable<SearchHit> hits = searchAll("index");
152-
assertThat(hits, everyItem(hasType("_doc")));
153+
assertThat(hits, everyItem(hasType("global_type")));
153154
}
154155

155156
@SuppressWarnings("unchecked")
156-
@AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/pull/36549")
157157
public void testTypeGlobalAndPerRequest() throws IOException {
158158
BulkRequest request = new BulkRequest(null, "global_type");
159159
request.add(new IndexRequest("index1", "local_type", "1")
160160
.source(XContentType.JSON, "field", "bulk1"));
161161
request.add(new IndexRequest("index2").id("2") // will take global type
162162
.source(XContentType.JSON, "field", "bulk2"));
163163

164-
bulk(request);
164+
bulkWithTypes(request);
165165

166166
Iterable<SearchHit> hits = searchAll("index1", "index2");
167167
assertThat(hits, containsInAnyOrder(
@@ -174,7 +174,7 @@ public void testTypeGlobalAndPerRequest() throws IOException {
174174
@SuppressWarnings("unchecked")
175175
public void testGlobalRouting() throws IOException {
176176
createIndexWithMultipleShards("index");
177-
BulkRequest request = new BulkRequest(null, null);
177+
BulkRequest request = new BulkRequest(null);
178178
request.add(new IndexRequest("index").id("1")
179179
.source(XContentType.JSON, "field", "bulk1"));
180180
request.add(new IndexRequest("index").id("2")
@@ -191,7 +191,7 @@ public void testGlobalRouting() throws IOException {
191191

192192
@SuppressWarnings("unchecked")
193193
public void testMixLocalAndGlobalRouting() throws IOException {
194-
BulkRequest request = new BulkRequest(null, null);
194+
BulkRequest request = new BulkRequest(null);
195195
request.routing("globalRouting");
196196
request.add(new IndexRequest("index").id("1")
197197
.source(XContentType.JSON, "field", "bulk1"));
@@ -204,12 +204,32 @@ public void testMixLocalAndGlobalRouting() throws IOException {
204204
Iterable<SearchHit> hits = searchAll(new SearchRequest("index").routing("globalRouting", "localRouting"));
205205
assertThat(hits, containsInAnyOrder(hasId("1"), hasId("2")));
206206
}
207+
208+
public void testGlobalIndexNoTypes() throws IOException {
209+
BulkRequest request = new BulkRequest("global_index");
210+
request.add(new IndexRequest().id("1")
211+
.source(XContentType.JSON, "field", "bulk1"));
212+
request.add(new IndexRequest().id("2")
213+
.source(XContentType.JSON, "field", "bulk2"));
207214

208-
private BulkResponse bulk(BulkRequest request) throws IOException {
209-
BulkResponse bulkResponse = execute(request, highLevelClient()::bulk, highLevelClient()::bulkAsync);
215+
bulk(request);
216+
217+
Iterable<SearchHit> hits = searchAll("global_index");
218+
assertThat(hits, everyItem(hasIndex("global_index")));
219+
}
220+
221+
private BulkResponse bulkWithTypes(BulkRequest request) throws IOException {
222+
BulkResponse bulkResponse = execute(request, highLevelClient()::bulk, highLevelClient()::bulkAsync,
223+
expectWarnings(RestBulkAction.TYPES_DEPRECATION_MESSAGE));
210224
assertFalse(bulkResponse.hasFailures());
211225
return bulkResponse;
212226
}
227+
228+
private BulkResponse bulk(BulkRequest request) throws IOException {
229+
BulkResponse bulkResponse = execute(request, highLevelClient()::bulk, highLevelClient()::bulkAsync, RequestOptions.DEFAULT);
230+
assertFalse(bulkResponse.hasFailures());
231+
return bulkResponse;
232+
}
213233

214234
@SuppressWarnings("unchecked")
215235
private static <T> Function<SearchHit, T> fieldFromSource(String fieldName) {

‎client/rest-high-level/src/test/java/org/elasticsearch/client/CrudIT.java‎

Lines changed: 5 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -65,6 +65,7 @@
6565
import org.elasticsearch.index.reindex.UpdateByQueryAction;
6666
import org.elasticsearch.index.reindex.UpdateByQueryRequest;
6767
import org.elasticsearch.rest.RestStatus;
68+
import org.elasticsearch.rest.action.document.RestBulkAction;
6869
import org.elasticsearch.rest.action.document.RestDeleteAction;
6970
import org.elasticsearch.rest.action.document.RestGetAction;
7071
import org.elasticsearch.rest.action.document.RestMultiGetAction;
@@ -449,7 +450,7 @@ public void testMultiGetWithTypes() throws IOException {
449450
bulk.add(new IndexRequest("index", "type", "id2")
450451
.source("{\"field\":\"value2\"}", XContentType.JSON));
451452

452-
highLevelClient().bulk(bulk, RequestOptions.DEFAULT);
453+
highLevelClient().bulk(bulk, expectWarnings(RestBulkAction.TYPES_DEPRECATION_MESSAGE));
453454
MultiGetRequest multiGetRequest = new MultiGetRequest();
454455
multiGetRequest.add("index", "id1");
455456
multiGetRequest.add("index", "type", "id2");
@@ -819,7 +820,7 @@ public void testBulk() throws IOException {
819820
}
820821
}
821822

822-
BulkResponse bulkResponse = execute(bulkRequest, highLevelClient()::bulk, highLevelClient()::bulkAsync);
823+
BulkResponse bulkResponse = execute(bulkRequest, highLevelClient()::bulk, highLevelClient()::bulkAsync, RequestOptions.DEFAULT);
823824
assertEquals(RestStatus.OK, bulkResponse.status());
824825
assertTrue(bulkResponse.getTook().getMillis() > 0);
825826
assertEquals(nbItems, bulkResponse.getItems().length);
@@ -1080,7 +1081,8 @@ public void afterBulk(long executionId, BulkRequest request, Throwable failure)
10801081
};
10811082

10821083
try (BulkProcessor processor = BulkProcessor.builder(
1083-
(request, bulkListener) -> highLevelClient().bulkAsync(request, RequestOptions.DEFAULT, bulkListener), listener)
1084+
(request, bulkListener) -> highLevelClient().bulkAsync(request,
1085+
RequestOptions.DEFAULT, bulkListener), listener)
10841086
.setConcurrentRequests(0)
10851087
.setBulkSize(new ByteSizeValue(5, ByteSizeUnit.GB))
10861088
.setBulkActions(nbItems + 1)

0 commit comments

Comments
 (0)