From 1c8cfc3a2b1fa5b6e2f53465233f60ce3cf65e5a Mon Sep 17 00:00:00 2001 From: Rishikesh1159 Date: Mon, 12 Feb 2024 21:47:37 +0000 Subject: [PATCH 1/3] Add new method to pick a random replication strategy. Signed-off-by: Rishikesh1159 --- .../opensearch/cluster/SimpleClusterStateIT.java | 7 ++++++- .../indices/IndicesRequestCacheIT.java | 9 +++++++-- .../opensearch/test/OpenSearchIntegTestCase.java | 16 ++++++++++++++++ 3 files changed, 29 insertions(+), 3 deletions(-) diff --git a/server/src/internalClusterTest/java/org/opensearch/cluster/SimpleClusterStateIT.java b/server/src/internalClusterTest/java/org/opensearch/cluster/SimpleClusterStateIT.java index f0337e9c0c84c..af5900b1cba6c 100644 --- a/server/src/internalClusterTest/java/org/opensearch/cluster/SimpleClusterStateIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/cluster/SimpleClusterStateIT.java @@ -97,12 +97,17 @@ protected Collection> nodePlugins() { return Collections.singletonList(PrivateCustomPlugin.class); } + @Override + protected boolean useRandomReplicationStrategy() { + return true; + } + @Before public void indexData() throws Exception { index("foo", "bar", "1", XContentFactory.jsonBuilder().startObject().field("foo", "foo").endObject()); index("fuu", "buu", "1", XContentFactory.jsonBuilder().startObject().field("fuu", "fuu").endObject()); index("baz", "baz", "1", XContentFactory.jsonBuilder().startObject().field("baz", "baz").endObject()); - refresh(); + refreshAndWaitForReplication(); } public void testRoutingTable() throws Exception { diff --git a/server/src/internalClusterTest/java/org/opensearch/indices/IndicesRequestCacheIT.java b/server/src/internalClusterTest/java/org/opensearch/indices/IndicesRequestCacheIT.java index 95f6a7a03266c..6c7e9cab2d4cc 100644 --- a/server/src/internalClusterTest/java/org/opensearch/indices/IndicesRequestCacheIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/indices/IndicesRequestCacheIT.java @@ -81,6 +81,11 @@ public static Collection parameters() { ); } + @Override + protected boolean useRandomReplicationStrategy() { + return true; + } + // One of the primary purposes of the query cache is to cache aggs results public void testCacheAggs() throws Exception { Client client = client(); @@ -554,7 +559,7 @@ public void testCacheWithFilteredAlias() throws InterruptedException { // Force merge the index to ensure there can be no background merges during the subsequent searches that would invalidate the cache ForceMergeResponse forceMergeResponse = client.admin().indices().prepareForceMerge("index").setFlush(true).get(); OpenSearchAssertions.assertAllSuccessful(forceMergeResponse); - refresh(); + refreshAndWaitForReplication(); indexRandomForConcurrentSearch("index"); @@ -661,7 +666,7 @@ public void testCacheWithInvalidation() throws Exception { assertCacheState(client, "index", 1, 1); // Explicit refresh would invalidate cache - refresh(); + refreshAndWaitForReplication(); // Hit same query again resp = client.prepareSearch("index").setRequestCache(true).setQuery(QueryBuilders.termQuery("k", "hello")).get(); assertSearchResponse(resp); diff --git a/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java b/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java index 33d5669d33297..0db9ef25e25ce 100644 --- a/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java +++ b/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java @@ -141,6 +141,7 @@ import org.opensearch.indices.IndicesQueryCache; import org.opensearch.indices.IndicesRequestCache; import org.opensearch.indices.IndicesService; +import org.opensearch.indices.replication.common.ReplicationType; import org.opensearch.indices.store.IndicesStore; import org.opensearch.monitor.os.OsInfo; import org.opensearch.node.NodeMocksPlugin; @@ -208,6 +209,7 @@ import static org.opensearch.index.IndexSettings.INDEX_DOC_ID_FUZZY_SET_FALSE_POSITIVE_PROBABILITY_SETTING; import static org.opensearch.index.IndexSettings.INDEX_SOFT_DELETES_RETENTION_LEASE_PERIOD_SETTING; import static org.opensearch.index.query.QueryBuilders.matchAllQuery; +import static org.opensearch.indices.IndicesService.CLUSTER_REPLICATION_TYPE_SETTING; import static org.opensearch.test.XContentTestUtils.convertToMap; import static org.opensearch.test.XContentTestUtils.differenceBetweenMapsIgnoringArrayOrder; import static org.opensearch.test.hamcrest.OpenSearchAssertions.assertAcked; @@ -1887,9 +1889,23 @@ protected Settings nodeSettings(int nodeOrdinal) { builder.put(TelemetrySettings.TRACER_FEATURE_ENABLED_SETTING.getKey(), true); builder.put(TelemetrySettings.TRACER_ENABLED_SETTING.getKey(), true); } + + // Randomly set a replication strategy for the node. + if (useRandomReplicationStrategy()) { + builder.put(CLUSTER_REPLICATION_TYPE_SETTING.getKey(), randomBoolean() ? ReplicationType.DOCUMENT : ReplicationType.SEGMENT); + } return builder.build(); } + /** + * Used for selecting random replication strategy, either DOCUMENT or SEGMENT. + * This method must be overridden by subclass to use random replication strategy. + * Should be used only on test classes where replication strategy is not critical for tests. + */ + protected boolean useRandomReplicationStrategy() { + return false; + } + protected Path nodeConfigPath(int nodeOrdinal) { return null; } From ba618e29c26576fd9e8b54ade2220823c7a026bf Mon Sep 17 00:00:00 2001 From: Rishikesh1159 Date: Mon, 12 Feb 2024 23:06:25 +0000 Subject: [PATCH 2/3] replace usage of refresh with refreshandWaitForReplication() Signed-off-by: Rishikesh1159 --- .../org/opensearch/indices/IndicesRequestCacheIT.java | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/server/src/internalClusterTest/java/org/opensearch/indices/IndicesRequestCacheIT.java b/server/src/internalClusterTest/java/org/opensearch/indices/IndicesRequestCacheIT.java index 6c7e9cab2d4cc..82577eb1501f3 100644 --- a/server/src/internalClusterTest/java/org/opensearch/indices/IndicesRequestCacheIT.java +++ b/server/src/internalClusterTest/java/org/opensearch/indices/IndicesRequestCacheIT.java @@ -185,7 +185,7 @@ public void testQueryRewrite() throws Exception { // Force merge the index to ensure there can be no background merges during the subsequent searches that would invalidate the cache ForceMergeResponse forceMergeResponse = client.admin().indices().prepareForceMerge("index").setFlush(true).get(); OpenSearchAssertions.assertAllSuccessful(forceMergeResponse); - refresh(); + refreshAndWaitForReplication(); ensureSearchable("index"); assertCacheState(client, "index", 0, 0); @@ -255,7 +255,7 @@ public void testQueryRewriteMissingValues() throws Exception { // Force merge the index to ensure there can be no background merges during the subsequent searches that would invalidate the cache ForceMergeResponse forceMergeResponse = client.admin().indices().prepareForceMerge("index").setFlush(true).get(); OpenSearchAssertions.assertAllSuccessful(forceMergeResponse); - refresh(); + refreshAndWaitForReplication(); ensureSearchable("index"); assertCacheState(client, "index", 0, 0); @@ -321,7 +321,7 @@ public void testQueryRewriteDates() throws Exception { // Force merge the index to ensure there can be no background merges during the subsequent searches that would invalidate the cache ForceMergeResponse forceMergeResponse = client.admin().indices().prepareForceMerge("index").setFlush(true).get(); OpenSearchAssertions.assertAllSuccessful(forceMergeResponse); - refresh(); + refreshAndWaitForReplication(); ensureSearchable("index"); assertCacheState(client, "index", 0, 0); @@ -394,7 +394,7 @@ public void testQueryRewriteDatesWithNow() throws Exception { .setFlush(true) .get(); OpenSearchAssertions.assertAllSuccessful(forceMergeResponse); - refresh(); + refreshAndWaitForReplication(); ensureSearchable("index-1", "index-2", "index-3"); assertCacheState(client, "index-1", 0, 0); @@ -465,7 +465,7 @@ public void testCanCache() throws Exception { // Force merge the index to ensure there can be no background merges during the subsequent searches that would invalidate the cache ForceMergeResponse forceMergeResponse = client.admin().indices().prepareForceMerge("index").setFlush(true).get(); OpenSearchAssertions.assertAllSuccessful(forceMergeResponse); - refresh(); + refreshAndWaitForReplication(); ensureSearchable("index"); assertCacheState(client, "index", 0, 0); From 6951ea220d85ec58ce0802c3d2dab912f1cfcb96 Mon Sep 17 00:00:00 2001 From: Rishikesh1159 Date: Wed, 14 Feb 2024 22:09:52 +0000 Subject: [PATCH 3/3] Add replication strategy logging. Signed-off-by: Rishikesh1159 --- .../java/org/opensearch/test/OpenSearchIntegTestCase.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java b/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java index 0db9ef25e25ce..47dd033834f1c 100644 --- a/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java +++ b/test/framework/src/main/java/org/opensearch/test/OpenSearchIntegTestCase.java @@ -1890,9 +1890,11 @@ protected Settings nodeSettings(int nodeOrdinal) { builder.put(TelemetrySettings.TRACER_ENABLED_SETTING.getKey(), true); } - // Randomly set a replication strategy for the node. + // Randomly set a replication strategy for the node. Replication Strategy can still be manually overridden by subclass if needed. if (useRandomReplicationStrategy()) { - builder.put(CLUSTER_REPLICATION_TYPE_SETTING.getKey(), randomBoolean() ? ReplicationType.DOCUMENT : ReplicationType.SEGMENT); + ReplicationType replicationType = randomBoolean() ? ReplicationType.DOCUMENT : ReplicationType.SEGMENT; + logger.info("Randomly using Replication Strategy as {}.", replicationType.toString()); + builder.put(CLUSTER_REPLICATION_TYPE_SETTING.getKey(), replicationType); } return builder.build(); }