From 62bcbdcdd59fd258b95efb9dfc0fa3e04f733922 Mon Sep 17 00:00:00 2001 From: Dimitrios Liappis Date: Mon, 23 Nov 2020 12:49:58 +0200 Subject: [PATCH] Remove implicit request timeout from force merge operation in polling Currently in polling mode, the force merge operation uses a very short timeout (1s) which, in case of slow response times, might actually cause the operation to not get executed. Since #1070 we can specify per operation request-timeout and we should honor that one instead. --- esrally/driver/runner.py | 5 +---- tests/driver/runner_test.py | 8 ++++---- 2 files changed, 5 insertions(+), 8 deletions(-) diff --git a/esrally/driver/runner.py b/esrally/driver/runner.py index 5cb8f9168..57f3bde34 100644 --- a/esrally/driver/runner.py +++ b/esrally/driver/runner.py @@ -676,9 +676,6 @@ async def __call__(self, es, params): if max_num_segments: merge_params["max_num_segments"] = max_num_segments if mode == "polling": - # we ignore the request_timeout if we are in polling mode and deliberately timeout early - # no reason to wait as long as a whole {polling-period} (which has a minimum of 1 second) - merge_params["request_timeout"] = 1 complete = False try: await es.indices.forcemerge(**merge_params) @@ -687,7 +684,7 @@ async def __call__(self, es, params): pass while not complete: await asyncio.sleep(params.get("poll-period")) - tasks = await es.tasks.list(params={"actions":"indices:admin/forcemerge"}) + tasks = await es.tasks.list(params={"actions": "indices:admin/forcemerge"}) if len(tasks["nodes"]) == 0: # empty nodes response indicates no tasks complete = True diff --git a/tests/driver/runner_test.py b/tests/driver/runner_test.py index 81f38d4b4..932f34892 100644 --- a/tests/driver/runner_test.py +++ b/tests/driver/runner_test.py @@ -988,7 +988,7 @@ async def test_force_merge_with_polling_no_timeout(self, es): force_merge = runner.ForceMerge() await force_merge(es, params={"index" : "_all", "mode": "polling", 'poll-period': 0}) - es.indices.forcemerge.assert_called_once_with(index="_all", request_timeout=1) + es.indices.forcemerge.assert_called_once_with(index="_all") @mock.patch("elasticsearch.Elasticsearch") @run_async @@ -1038,8 +1038,8 @@ async def test_force_merge_with_polling(self, es): }) ] force_merge = runner.ForceMerge() - await force_merge(es, params={"index" : "_all", "mode": "polling", "poll-period": 0}) - es.indices.forcemerge.assert_called_once_with(index="_all", request_timeout=1) + await force_merge(es, params={"index": "_all", "mode": "polling", "poll-period": 0}) + es.indices.forcemerge.assert_called_once_with(index="_all") @mock.patch("elasticsearch.Elasticsearch") @run_async @@ -1092,7 +1092,7 @@ async def test_force_merge_with_polling_and_params(self, es): # request-timeout should be ignored as mode:polling await force_merge(es, params={"index" : "_all", "mode": "polling", "max-num-segments": 1, "request-timeout": 50000, "poll-period": 0}) - es.indices.forcemerge.assert_called_once_with(index="_all", max_num_segments=1, request_timeout=1) + es.indices.forcemerge.assert_called_once_with(index="_all", max_num_segments=1, request_timeout=50000) class IndicesStatsRunnerTests(TestCase):