From f73aac93b78691dbf197d3fd35f7406fa8717e9c Mon Sep 17 00:00:00 2001 From: Suraj Singh Date: Fri, 11 Aug 2023 14:38:18 -0700 Subject: [PATCH] [Segment Replication] Unmute testIndexingWithSegRep rolling upgrade test (#9079) * Unmute testIndexingWithSegRep rolling upgrade test Signed-off-by: Suraj Singh * Assert on row count before processing the result Signed-off-by: Suraj Singh * Filter rows with 0 doc count Signed-off-by: Suraj Singh * Retry assertHitCount Signed-off-by: Suraj Singh * Handle exception Signed-off-by: Suraj Singh * Add comment Signed-off-by: Suraj Singh --------- Signed-off-by: Suraj Singh --- .../org/opensearch/upgrades/IndexingIT.java | 36 ++++++++++++------- 1 file changed, 24 insertions(+), 12 deletions(-) diff --git a/qa/rolling-upgrade/src/test/java/org/opensearch/upgrades/IndexingIT.java b/qa/rolling-upgrade/src/test/java/org/opensearch/upgrades/IndexingIT.java index a03d299b32274..187ec19fb9ef7 100644 --- a/qa/rolling-upgrade/src/test/java/org/opensearch/upgrades/IndexingIT.java +++ b/qa/rolling-upgrade/src/test/java/org/opensearch/upgrades/IndexingIT.java @@ -106,14 +106,25 @@ private void waitForSearchableDocs(String index, int shardCount, int replicaCoun Request segrepStatsRequest = new Request("GET", "/_cat/segments/" + index + "?s=shard,segment,primaryOrReplica"); segrepStatsRequest.addParameter("h", "index,shard,primaryOrReplica,segment,docs.count"); Response segrepStatsResponse = client().performRequest(segrepStatsRequest); - logger.info("--> _cat/segments response\n {}", EntityUtils.toString(segrepStatsResponse.getEntity())); List responseList = Streams.readAllLines(segrepStatsResponse.getEntity().getContent()); - for (int segmentsIndex=0; segmentsIndex < responseList.size();) { - String[] primaryRow = responseList.get(segmentsIndex++).split(" +"); + logger.info("--> _cat/segments response\n {}", responseList.toString().replace(',', '\n')); + // Filter response for rows with zero doc count + List filteredList = new ArrayList<>(); + for(String row: responseList) { + String count = row.split(" +")[4]; + if (count.equals("0") == false) { + filteredList.add(row); + } + } + // Ensure there is result for replica copies before processing the result. This results in retry when there + // are not enough number of rows vs failing with IndexOutOfBoundsException + assertEquals(0, filteredList.size() % (replicaCount + 1)); + for (int segmentsIndex=0; segmentsIndex < filteredList.size();) { + String[] primaryRow = filteredList.get(segmentsIndex++).split(" +"); String shardId = primaryRow[0] + primaryRow[1]; assertTrue(primaryRow[2].equals("p")); for(int replicaIndex = 1; replicaIndex <= replicaCount; replicaIndex++) { - String[] replicaRow = responseList.get(segmentsIndex).split(" +"); + String[] replicaRow = filteredList.get(segmentsIndex).split(" +"); String replicaShardId = replicaRow[0] + replicaRow[1]; // When segment has 0 doc count, not all replica copies posses that segment. Skip to next segment if (replicaRow[2].equals("p")) { @@ -157,7 +168,7 @@ private void verifySegmentStats(String indexName) throws Exception { }, 1, TimeUnit.MINUTES); } - public void testIndexing() throws IOException, ParseException { + public void testIndexing() throws Exception { switch (CLUSTER_TYPE) { case OLD: break; @@ -250,7 +261,6 @@ public void testIndexing() throws IOException, ParseException { * * @throws Exception */ - @AwaitsFix(bugUrl = "https://github.com/opensearch-project/OpenSearch/issues/8322") public void testIndexingWithSegRep() throws Exception { if (UPGRADE_FROM_VERSION.before(Version.V_2_4_0)) { logger.info("--> Skip test for version {} where segment replication feature is not available", UPGRADE_FROM_VERSION); @@ -389,12 +399,14 @@ private void bulk(String index, String valueSuffix, int count) throws IOExceptio client().performRequest(bulk); } - private void assertCount(String index, int count) throws IOException, ParseException { - Request searchTestIndexRequest = new Request("POST", "/" + index + "/_search"); - searchTestIndexRequest.addParameter(TOTAL_HITS_AS_INT_PARAM, "true"); - searchTestIndexRequest.addParameter("filter_path", "hits.total"); - Response searchTestIndexResponse = client().performRequest(searchTestIndexRequest); - assertEquals("{\"hits\":{\"total\":" + count + "}}", + private void assertCount(String index, int count) throws Exception { + assertBusy(() -> { + Request searchTestIndexRequest = new Request("POST", "/" + index + "/_search"); + searchTestIndexRequest.addParameter(TOTAL_HITS_AS_INT_PARAM, "true"); + searchTestIndexRequest.addParameter("filter_path", "hits.total"); + Response searchTestIndexResponse = client().performRequest(searchTestIndexRequest); + assertEquals("{\"hits\":{\"total\":" + count + "}}", EntityUtils.toString(searchTestIndexResponse.getEntity(), StandardCharsets.UTF_8)); + }, 30, TimeUnit.SECONDS); } }