From 3d62775e886f4f3d2fbf5654b027e18d5fb36189 Mon Sep 17 00:00:00 2001 From: Dawid Wysakowicz Date: Tue, 2 Jun 2020 08:51:35 +0200 Subject: [PATCH] [FLINK-18052] Increase timeout for ES Search API in IT Cases This closes #12434 --- .../table/Elasticsearch6DynamicSinkITCase.java | 12 ++++++++---- .../table/Elasticsearch7DynamicSinkITCase.java | 12 ++++++++---- 2 files changed, 16 insertions(+), 8 deletions(-) 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 3c096532abc3d..41a67f62aae0a 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 @@ 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 result = hits.getAt(0).getSourceAsMap(); Map 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 3b667dc13aecb..dbd1ff9031fe0 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 @@ 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 result = hits.getAt(0).getSourceAsMap(); Map expectedMap = new HashMap<>();