[FLINK-18052] Increase timeout for ES Search API in IT Cases

This closes apache#12434
zhangjun0x01 · Jul 7, 2020 · 3d62775 · 3d62775
1 parent 45f4cf1
commit 3d62775
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 8 deletions.
diff --git a/...pache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6DynamicSinkITCase.java b/...pache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6DynamicSinkITCase.java
@@ -219,17 +219,21 @@ public void testWritingDocumentsNoPrimaryKey() throws Exception {
 
 		// search API does not return documents that were not indexed, we might need to query
 		// the index a few times
-		Deadline deadline = Deadline.fromNow(Duration.ofSeconds(1));
+		Deadline deadline = Deadline.fromNow(Duration.ofSeconds(30));
 		SearchHits hits;
 		do {
 			hits = client.prepareSearch(index)
 				.execute()
 				.actionGet()
 				.getHits();
-			if (hits.getTotalHits() == 0) {
-				Thread.sleep(100);
+			if (hits.getTotalHits() < 1) {
+				Thread.sleep(200);
 			}
-		} while (hits.getTotalHits() == 0 && deadline.hasTimeLeft());
+		} while (hits.getTotalHits() < 1 && deadline.hasTimeLeft());
+
+		if (hits.getTotalHits() < 1) {
+			throw new AssertionError("Could not retrieve results from Elasticsearch.");
+		}
 
 		Map<String, Object> result = hits.getAt(0).getSourceAsMap();
 		Map<Object, Object> expectedMap = new HashMap<>();

diff --git a/...pache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSinkITCase.java b/...pache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSinkITCase.java
@@ -211,17 +211,21 @@ public void testWritingDocumentsNoPrimaryKey() throws Exception {
 
 		// search API does not return documents that were not indexed, we might need to query
 		// the index a few times
-		Deadline deadline = Deadline.fromNow(Duration.ofSeconds(1));
+		Deadline deadline = Deadline.fromNow(Duration.ofSeconds(30));
 		SearchHits hits;
 		do {
 			hits = client.prepareSearch(index)
 				.execute()
 				.actionGet()
 				.getHits();
-			if (hits.getTotalHits().value == 0) {
-				Thread.sleep(100);
+			if (hits.getTotalHits().value < 1) {
+				Thread.sleep(200);
 			}
-		} while (hits.getTotalHits().value == 0 && deadline.hasTimeLeft());
+		} while (hits.getTotalHits().value < 1 && deadline.hasTimeLeft());
+
+		if (hits.getTotalHits().value < 1) {
+			throw new AssertionError("Could not retrieve results from Elasticsearch.");
+		}
 
 		Map<String, Object> result = hits.getAt(0).getSourceAsMap();
 		Map<Object, Object> expectedMap = new HashMap<>();