[FLINK-18052] Increase timeout for ES Search API in IT Cases
This closes #12434
diff --git a/flink-connectors/flink-connector-elasticsearch6/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6DynamicSinkITCase.java b/flink-connectors/flink-connector-elasticsearch6/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6DynamicSinkITCase.java
index 3c09653..41a67f6 100644
--- a/flink-connectors/flink-connector-elasticsearch6/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6DynamicSinkITCase.java
+++ b/flink-connectors/flink-connector-elasticsearch6/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch6DynamicSinkITCase.java
@@ -219,17 +219,21 @@
// 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/flink-connectors/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSinkITCase.java b/flink-connectors/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSinkITCase.java
index 3b667dc..dbd1ff9 100644
--- a/flink-connectors/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSinkITCase.java
+++ b/flink-connectors/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/table/Elasticsearch7DynamicSinkITCase.java
@@ -211,17 +211,21 @@
// 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<>();
