From 26d6e27f32c252ebb4011fdc7b5fab47b0452b2b Mon Sep 17 00:00:00 2001 From: Chen Dai Date: Wed, 22 Nov 2023 15:19:00 -0800 Subject: [PATCH] Reword user manual Signed-off-by: Chen Dai --- docs/index.md | 6 +++--- .../flint/core/http/RetryableHttpAsyncClient.java | 13 +++++++++---- 2 files changed, 12 insertions(+), 7 deletions(-) diff --git a/docs/index.md b/docs/index.md index 3c3f02dc5..03164e942 100644 --- a/docs/index.md +++ b/docs/index.md @@ -358,9 +358,9 @@ In the index mapping, the `_meta` and `properties`field stores meta and schema i (false), IMMEDIATE(true), WAIT_UNTIL(wait_for)] - `spark.datasource.flint.read.scroll_size`: default value is 100. - `spark.datasource.flint.read.scroll_duration`: default value is 5 minutes. scroll context keep alive duration. -- `spark.datasource.flint.retry.max_retries`: default value is 3. max retries on failed HTTP request. 0 means retry is disabled. -- `spark.datasource.flint.retry.http_status_codes`: default value is "429,502". retryable HTTP response status code list -- `spark.datasource.flint.retry.exception_class_names`: by default no retry on any exception thrown. retryable exception class name list. +- `spark.datasource.flint.retry.max_retries`: max retries on failed HTTP request. default value is 3. Use 0 to disable retry. +- `spark.datasource.flint.retry.http_status_codes`: retryable HTTP response status code list. default value is "429,502" (429 Too Many Request and 502 Bad Gateway). +- `spark.datasource.flint.retry.exception_class_names`: retryable exception class name list. by default no retry on any exception thrown. - `spark.flint.optimizer.enabled`: default is true. - `spark.flint.index.hybridscan.enabled`: default is false. - `spark.flint.index.checkpoint.mandatory`: default is true. diff --git a/flint-core/src/main/scala/org/opensearch/flint/core/http/RetryableHttpAsyncClient.java b/flint-core/src/main/scala/org/opensearch/flint/core/http/RetryableHttpAsyncClient.java index 2bc248e04..c7c9258ec 100644 --- a/flint-core/src/main/scala/org/opensearch/flint/core/http/RetryableHttpAsyncClient.java +++ b/flint-core/src/main/scala/org/opensearch/flint/core/http/RetryableHttpAsyncClient.java @@ -65,7 +65,12 @@ public Future execute(HttpAsyncRequestProducer requestProducer, HttpContext context, FutureCallback callback) { return new Future<>() { - /** Delegate future object created per execution */ + /** + * Delegated future object created per doExecuteAndFutureGetWithRetry() call which creates initial object too. + * In this way, we avoid the duplicate logic of first call and subsequent retry calls. + * Here the assumption is cancel, isCancelled and isDone never called before get(). + * (OpenSearch RestClient seems only call get() API) + */ private Future delegate; @Override @@ -85,15 +90,15 @@ public boolean isDone() { @Override public T get() throws InterruptedException, ExecutionException { - return doGetWithRetry(() -> delegate.get()); + return doExecuteAndFutureGetWithRetry(() -> delegate.get()); } @Override public T get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException { - return doGetWithRetry(() -> delegate.get(timeout, unit)); + return doExecuteAndFutureGetWithRetry(() -> delegate.get(timeout, unit)); } - private T doGetWithRetry(Callable futureGet) throws InterruptedException, ExecutionException { + private T doExecuteAndFutureGetWithRetry(Callable futureGet) throws InterruptedException, ExecutionException { try { // Retry by creating a new Future object (as new delegate) and get its result again return Failsafe