diff --git a/hetu-docs/en/admin/properties.md b/hetu-docs/en/admin/properties.md index 966260256..8e9bc45b5 100644 --- a/hetu-docs/en/admin/properties.md +++ b/hetu-docs/en/admin/properties.md @@ -389,7 +389,7 @@ Exchanges transfer data between openLooKeng nodes for different stages of a quer > - **Type:** `integer` > - **Default value:** `10` > -> The maximum number of retry for failed task performed by the coordinator before considering it as a permanent failure. This property is used only when exchange.is-timeout-failure-detection-enabled is set to false. +> The maximum number of retry for failed task performed by the coordinator before considering it as a permanent failure. This property is used only when exchange.is-timeout-failure-detection-enabled is set to false. This value needs to be atleast 3 (minimum retry count) to take effect. ### `sink.max-buffer-size` diff --git a/presto-main/src/main/java/io/prestosql/server/remotetask/Backoff.java b/presto-main/src/main/java/io/prestosql/server/remotetask/Backoff.java index f66b0385d..ef2cde01e 100644 --- a/presto-main/src/main/java/io/prestosql/server/remotetask/Backoff.java +++ b/presto-main/src/main/java/io/prestosql/server/remotetask/Backoff.java @@ -75,14 +75,13 @@ public class Backoff public Backoff(int minTries, Duration maxFailureInterval, Ticker ticker, List backoffDelayIntervals, int maxTries) { checkArgument(minTries > 0, "minTries must be at least 1"); - checkArgument(maxTries > 0, "maxTries must be at least 1"); requireNonNull(maxFailureInterval, "maxFailureInterval is null"); requireNonNull(ticker, "ticker is null"); requireNonNull(backoffDelayIntervals, "backoffDelayIntervals is null"); checkArgument(!backoffDelayIntervals.isEmpty(), "backoffDelayIntervals must contain at least one entry"); this.minTries = minTries; - this.maxTries = maxTries; + this.maxTries = (minTries < maxTries) ? maxTries : minTries; this.maxFailureIntervalNanos = maxFailureInterval.roundTo(NANOSECONDS); this.ticker = ticker; this.backoffDelayIntervalsNanos = backoffDelayIntervals.stream() @@ -100,7 +99,7 @@ public class Backoff checkArgument(!backoffDelayIntervals.isEmpty(), "backoffDelayIntervals must contain at least one entry"); this.minTries = minTries; - this.maxTries = minTries + MAX_RETRIES; // to guaranty higher max retry value when min retry is > MAX_RETRIES + this.maxTries = (minTries < MAX_RETRIES) ? MAX_RETRIES : minTries; this.maxFailureIntervalNanos = maxFailureInterval.roundTo(NANOSECONDS); this.ticker = ticker; this.backoffDelayIntervalsNanos = backoffDelayIntervals.stream() @@ -136,10 +135,19 @@ public class Backoff { lastRequestStart = 0; firstFailureTime = 0; - failureCount = 0; + setFailureCount(0, false); lastFailureTime = 0; } + private synchronized void setFailureCount(int n, boolean isInc) + { + if (isInc) { + failureCount = failureCount + n; + return; + } + failureCount = n; + } + /** * @return true if max retry failed, now it is time to check node status from HeartbeatFailureDetector */ @@ -148,7 +156,7 @@ public class Backoff long now = ticker.read(); lastFailureTime = now; - failureCount++; + setFailureCount(1, true); if (lastRequestStart != 0) { failureRequestTimeTotal += now - lastRequestStart; lastRequestStart = 0; @@ -160,11 +168,10 @@ public class Backoff return false; } - if (failureCount < minTries) { + if (getFailureCount() < minTries) { return false; } - - return failureCount > maxTries; + return getFailureCount() >= maxTries; } /** @@ -186,7 +193,7 @@ public class Backoff long now = ticker.read(); lastFailureTime = now; - failureCount++; + setFailureCount(1, true); if (lastRequestStart != 0) { failureRequestTimeTotal += now - lastRequestStart; lastRequestStart = 0; @@ -198,7 +205,7 @@ public class Backoff return false; } - if (failureCount < minTries) { + if (getFailureCount() < minTries) { return false; } @@ -208,7 +215,7 @@ public class Backoff public synchronized long getBackoffDelayNanos() { - int tmpFailureCount = (int) min(backoffDelayIntervalsNanos.length, this.failureCount); + int tmpFailureCount = (int) min(backoffDelayIntervalsNanos.length, getFailureCount()); if (tmpFailureCount == 0) { return 0; } diff --git a/presto-main/src/test/java/io/prestosql/operator/TestHttpPageBufferClient.java b/presto-main/src/test/java/io/prestosql/operator/TestHttpPageBufferClient.java index c9c0edc77..5a8b67deb 100644 --- a/presto-main/src/test/java/io/prestosql/operator/TestHttpPageBufferClient.java +++ b/presto-main/src/test/java/io/prestosql/operator/TestHttpPageBufferClient.java @@ -459,9 +459,9 @@ public class TestHttpPageBufferClient assertEquals(callback.getPages().size(), 0); assertEquals(callback.getCompletedRequests(), 11); assertEquals(callback.getFinishedBuffers(), 0); - assertEquals(callback.getFailedBuffers(), 1); + assertEquals(callback.getFailedBuffers(), 2); assertInstanceOf(callback.getFailure(), PageTransportTimeoutException.class); - assertContains(callback.getFailure().getMessage(), WORKER_NODE_ERROR + " (http://localhost:8080/0 - 11 failures,"); + assertContains(callback.getFailure().getMessage(), WORKER_NODE_ERROR + " (http://localhost:8080/0 - 10 failures,"); assertStatus(client, location, "queued", 0, 11, 11, 11, "not scheduled"); }