|
19 | 19 | import org.opensearch.cluster.metadata.IndexMetadata; |
20 | 20 | import org.opensearch.cluster.routing.allocation.command.AllocateReplicaAllocationCommand; |
21 | 21 | import org.opensearch.common.settings.Settings; |
| 22 | +import org.opensearch.index.query.BoolQueryBuilder; |
22 | 23 | import org.opensearch.index.query.RangeQueryBuilder; |
| 24 | +import org.opensearch.index.query.TermQueryBuilder; |
23 | 25 | import org.opensearch.test.InternalTestCluster; |
24 | 26 | import org.opensearch.test.OpenSearchIntegTestCase; |
25 | 27 | import org.opensearch.transport.client.Requests; |
@@ -310,6 +312,122 @@ public void testPaginatedGetIngestionState() throws ExecutionException, Interrup |
310 | 312 | })); |
311 | 313 | } |
312 | 314 |
|
| 315 | + public void testExternalVersioning() throws Exception { |
| 316 | + // setup nodes and index |
| 317 | + produceDataWithExternalVersion("1", 1, "name1", "25", defaultMessageTimestamp, "index"); |
| 318 | + produceDataWithExternalVersion("2", 1, "name2", "25", defaultMessageTimestamp, "index"); |
| 319 | + internalCluster().startClusterManagerOnlyNode(); |
| 320 | + final String nodeA = internalCluster().startDataOnlyNode(); |
| 321 | + final String nodeB = internalCluster().startDataOnlyNode(); |
| 322 | + |
| 323 | + createIndexWithDefaultSettings(1, 1); |
| 324 | + ensureGreen(indexName); |
| 325 | + waitForSearchableDocs(2, Arrays.asList(nodeA, nodeB)); |
| 326 | + |
| 327 | + // validate next version docs get indexed |
| 328 | + produceDataWithExternalVersion("1", 2, "name1", "30", defaultMessageTimestamp, "index"); |
| 329 | + produceDataWithExternalVersion("2", 2, "name2", "30", defaultMessageTimestamp, "index"); |
| 330 | + waitForState(() -> { |
| 331 | + BoolQueryBuilder query1 = new BoolQueryBuilder().must(new TermQueryBuilder("_id", 1)); |
| 332 | + SearchResponse response1 = client().prepareSearch(indexName).setQuery(query1).get(); |
| 333 | + assertThat(response1.getHits().getTotalHits().value(), is(1L)); |
| 334 | + BoolQueryBuilder query2 = new BoolQueryBuilder().must(new TermQueryBuilder("_id", 2)); |
| 335 | + SearchResponse response2 = client().prepareSearch(indexName).setQuery(query2).get(); |
| 336 | + assertThat(response2.getHits().getTotalHits().value(), is(1L)); |
| 337 | + return 30 == (Integer) response1.getHits().getHits()[0].getSourceAsMap().get("age") |
| 338 | + && 30 == (Integer) response2.getHits().getHits()[0].getSourceAsMap().get("age"); |
| 339 | + }); |
| 340 | + |
| 341 | + // test out-of-order updates |
| 342 | + produceDataWithExternalVersion("1", 1, "name1", "25", defaultMessageTimestamp, "index"); |
| 343 | + produceDataWithExternalVersion("2", 1, "name2", "25", defaultMessageTimestamp, "index"); |
| 344 | + produceDataWithExternalVersion("3", 1, "name3", "25", defaultMessageTimestamp, "index"); |
| 345 | + waitForSearchableDocs(3, Arrays.asList(nodeA, nodeB)); |
| 346 | + |
| 347 | + BoolQueryBuilder query1 = new BoolQueryBuilder().must(new TermQueryBuilder("_id", 1)); |
| 348 | + SearchResponse response1 = client().prepareSearch(indexName).setQuery(query1).get(); |
| 349 | + assertThat(response1.getHits().getTotalHits().value(), is(1L)); |
| 350 | + assertEquals(30, response1.getHits().getHits()[0].getSourceAsMap().get("age")); |
| 351 | + |
| 352 | + BoolQueryBuilder query2 = new BoolQueryBuilder().must(new TermQueryBuilder("_id", 2)); |
| 353 | + SearchResponse response2 = client().prepareSearch(indexName).setQuery(query2).get(); |
| 354 | + assertThat(response2.getHits().getTotalHits().value(), is(1L)); |
| 355 | + assertEquals(30, response2.getHits().getHits()[0].getSourceAsMap().get("age")); |
| 356 | + |
| 357 | + // test deletes with smaller version |
| 358 | + produceDataWithExternalVersion("1", 1, "name1", "25", defaultMessageTimestamp, "delete"); |
| 359 | + produceDataWithExternalVersion("4", 1, "name4", "25", defaultMessageTimestamp, "index"); |
| 360 | + waitForSearchableDocs(4, Arrays.asList(nodeA, nodeB)); |
| 361 | + RangeQueryBuilder query = new RangeQueryBuilder("age").gte(23); |
| 362 | + SearchResponse response = client().prepareSearch(indexName).setQuery(query).get(); |
| 363 | + assertThat(response.getHits().getTotalHits().value(), is(4L)); |
| 364 | + |
| 365 | + // test deletes with correct version |
| 366 | + produceDataWithExternalVersion("1", 3, "name1", "30", defaultMessageTimestamp, "delete"); |
| 367 | + produceDataWithExternalVersion("2", 3, "name2", "30", defaultMessageTimestamp, "delete"); |
| 368 | + waitForState(() -> { |
| 369 | + RangeQueryBuilder rangeQuery = new RangeQueryBuilder("age").gte(23); |
| 370 | + SearchResponse rangeQueryResponse = client().prepareSearch(indexName).setQuery(rangeQuery).get(); |
| 371 | + assertThat(rangeQueryResponse.getHits().getTotalHits().value(), is(2L)); |
| 372 | + return true; |
| 373 | + }); |
| 374 | + } |
| 375 | + |
| 376 | + public void testExternalVersioningWithDisabledGCDeletes() throws Exception { |
| 377 | + // setup nodes and index |
| 378 | + internalCluster().startClusterManagerOnlyNode(); |
| 379 | + final String nodeA = internalCluster().startDataOnlyNode(); |
| 380 | + final String nodeB = internalCluster().startDataOnlyNode(); |
| 381 | + |
| 382 | + createIndex( |
| 383 | + indexName, |
| 384 | + Settings.builder() |
| 385 | + .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1) |
| 386 | + .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 1) |
| 387 | + .put("ingestion_source.type", "kafka") |
| 388 | + .put("ingestion_source.pointer.init.reset", "earliest") |
| 389 | + .put("ingestion_source.param.topic", topicName) |
| 390 | + .put("ingestion_source.param.bootstrap_servers", kafka.getBootstrapServers()) |
| 391 | + .put("index.replication.type", "SEGMENT") |
| 392 | + .put("index.gc_deletes", "0") |
| 393 | + .build(), |
| 394 | + "{\"properties\":{\"name\":{\"type\": \"text\"},\"age\":{\"type\": \"integer\"}}}}" |
| 395 | + ); |
| 396 | + |
| 397 | + // insert documents |
| 398 | + produceDataWithExternalVersion("1", 1, "name1", "25", defaultMessageTimestamp, "index"); |
| 399 | + produceDataWithExternalVersion("2", 1, "name2", "25", defaultMessageTimestamp, "index"); |
| 400 | + waitForState(() -> { |
| 401 | + RangeQueryBuilder rangeQuery = new RangeQueryBuilder("age").gte(23); |
| 402 | + SearchResponse rangeQueryResponse = client().prepareSearch(indexName).setQuery(rangeQuery).get(); |
| 403 | + assertThat(rangeQueryResponse.getHits().getTotalHits().value(), is(2L)); |
| 404 | + return true; |
| 405 | + }); |
| 406 | + |
| 407 | + // delete documents 1 and 2 |
| 408 | + produceDataWithExternalVersion("1", 2, "name1", "25", defaultMessageTimestamp, "delete"); |
| 409 | + produceDataWithExternalVersion("2", 2, "name2", "25", defaultMessageTimestamp, "delete"); |
| 410 | + produceDataWithExternalVersion("3", 1, "name3", "25", defaultMessageTimestamp, "index"); |
| 411 | + waitForState(() -> { |
| 412 | + BoolQueryBuilder query = new BoolQueryBuilder().must(new TermQueryBuilder("_id", 3)); |
| 413 | + SearchResponse response = client().prepareSearch(indexName).setQuery(query).get(); |
| 414 | + assertThat(response.getHits().getTotalHits().value(), is(1L)); |
| 415 | + return 25 == (Integer) response.getHits().getHits()[0].getSourceAsMap().get("age"); |
| 416 | + }); |
| 417 | + waitForSearchableDocs(1, Arrays.asList(nodeA, nodeB)); |
| 418 | + |
| 419 | + // validate index operation with lower version creates new document |
| 420 | + produceDataWithExternalVersion("1", 1, "name1", "35", defaultMessageTimestamp, "index"); |
| 421 | + produceDataWithExternalVersion("4", 1, "name4", "35", defaultMessageTimestamp, "index"); |
| 422 | + waitForState(() -> { |
| 423 | + RangeQueryBuilder rangeQuery = new RangeQueryBuilder("age").gte(34); |
| 424 | + SearchResponse rangeQueryResponse = client().prepareSearch(indexName).setQuery(rangeQuery).get(); |
| 425 | + assertThat(rangeQueryResponse.getHits().getTotalHits().value(), is(2L)); |
| 426 | + return true; |
| 427 | + }); |
| 428 | + |
| 429 | + } |
| 430 | + |
313 | 431 | private void verifyRemoteStoreEnabled(String node) { |
314 | 432 | GetSettingsResponse settingsResponse = client(node).admin().indices().prepareGetSettings(indexName).get(); |
315 | 433 | String remoteStoreEnabled = settingsResponse.getIndexToSettings().get(indexName).get("index.remote_store.enabled"); |
|
0 commit comments