From 920bbbdf5faa0f6733e5667b1d3de4abcfdee915 Mon Sep 17 00:00:00 2001 From: Sebastian Gruza Date: Thu, 10 Sep 2026 01:25:39 +0000 Subject: [PATCH 1/5] fix(store-client): stop retrying a commit after DEADLINE_EXCEEDED and honour thread interrupts NodeTxExecutor.retryingInvoke() retried every failure up to NODE_MAX_RETRYING_TIMES (10) with a sleep schedule of 1,1,1,2..8 s and swallowed the InterruptedException from Thread.sleep. When a partition leader stops answering, one commit therefore held the calling REST worker for 11 x grpc.timeout.seconds + 38 s (about 19 min on defaults) and restserver.request_timeout could not stop it, so a single stalled store node exhausted the REST worker pool. Now the loop aborts when the calling thread is interrupted (restoring the interrupt flag) and does not retry a DEADLINE_EXCEEDED or CANCELLED status: a second attempt would only wait the full deadline again. UNAVAILABLE and other transport errors are retried as before (store replacement, leader change). --- .../store/client/NodeTxExecutor.java | 75 ++++++++++++++----- 1 file changed, 57 insertions(+), 18 deletions(-) diff --git a/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutor.java b/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutor.java index 9393421821..4305677302 100644 --- a/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutor.java +++ b/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutor.java @@ -46,6 +46,8 @@ import org.apache.hugegraph.store.term.HgPair; import org.apache.hugegraph.store.term.HgTriple; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; import lombok.extern.slf4j.Slf4j; /** @@ -376,32 +378,44 @@ Optional retryingInvoke(Supplier supplier) { return IntStream.rangeClosed(0, NODE_MAX_RETRYING_TIMES).boxed() .map( i -> { + if (Thread.currentThread().isInterrupted()) { + // The caller (e.g. a REST worker hitting + // restserver.request_timeout) gave up: stop + // retrying instead of holding its thread. + throw HgStoreClientException.of( + "Interrupted before retry " + i); + } T buffer = null; try { buffer = supplier.get(); } catch (Throwable t) { - if (i + 1 <= NODE_MAX_RETRYING_TIMES) { - try { - int sleepTime; - // The first three times try once every second - if (i < 3) { - sleepTime = 1; - } else { - // Subsequent incremental - sleepTime = i - 1; - } - log.info("Waiting {} seconds " + - "for the next try.", - sleepTime); - Thread.sleep(sleepTime * 1000L); - } catch (InterruptedException e) { - log.error("Failed to sleep", e); - } - } else { + if (i + 1 > NODE_MAX_RETRYING_TIMES) { log.error(maxTryMsg, t); throw HgStoreClientException.of( t.getMessage(), t); } + if (!isRetryable(t)) { + // A deadline or a cancellation will not + // get better by waiting the full deadline + // again; fail fast and let the caller decide. + log.warn("Not retrying after: {}", + t.getMessage()); + throw HgStoreClientException.of( + t.getMessage(), t); + } + // The first three times try once every second, + // subsequent incremental + int sleepTime = i < 3 ? 1 : i - 1; + log.info("Waiting {} seconds for the next try.", + sleepTime); + try { + Thread.sleep(sleepTime * 1000L); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw HgStoreClientException.of( + "Interrupted while waiting to retry: " + + t.getMessage(), t); + } } return buffer; } @@ -411,6 +425,31 @@ Optional retryingInvoke(Supplier supplier) { } + /** + * Retry only failures that a fresh attempt can plausibly fix (a transport error, a + * leader change). A DEADLINE_EXCEEDED would wait the full deadline again, a CANCELLED + * means the caller's thread was interrupted; neither is retried. + */ + static boolean isRetryable(Throwable t) { + Throwable c = t; + while (c != null) { + if (c instanceof InterruptedException) { + return false; + } + if (c instanceof StatusRuntimeException) { + Status.Code code = ((StatusRuntimeException) c).getStatus().getCode(); + if (code == Status.Code.DEADLINE_EXCEEDED || code == Status.Code.CANCELLED) { + return false; + } + } + if (c.getCause() == c) { + break; + } + c = c.getCause(); + } + return true; + } + private boolean isValid(Object obj) { if (obj == null) { return false; From 4350ef993bdc70cfb9fdbc9fdea9762cd3f37b3f Mon Sep 17 00:00:00 2001 From: Sebastian Gruza Date: Thu, 10 Sep 2026 15:18:48 +0000 Subject: [PATCH 2/5] test(store-client): cover retry classification, deadline fail-fast and interrupt handling in NodeTxExecutor --- .../store/client/NodeTxExecutorTest.java | 71 +++++++++++++++++++ 1 file changed, 71 insertions(+) diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutorTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutorTest.java index 00a05c65c3..629c52c756 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutorTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutorTest.java @@ -17,7 +17,11 @@ package org.apache.hugegraph.store.client; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -32,8 +36,12 @@ import java.util.concurrent.atomic.AtomicInteger; import org.apache.hugegraph.store.HgStoreSession; +import org.apache.hugegraph.store.client.type.HgStoreClientException; import org.junit.Test; +import io.grpc.Status; +import io.grpc.StatusRuntimeException; + public class NodeTxExecutorTest { @Test @@ -115,4 +123,67 @@ public void testParallelReplacementUsesOneCurrentSession() throws Exception { workers.shutdownNow(); } } + + @Test + public void testIsRetryableClassifiesFailures() { + assertFalse(NodeTxExecutor.isRetryable(Status.DEADLINE_EXCEEDED.asRuntimeException())); + assertFalse(NodeTxExecutor.isRetryable(Status.CANCELLED.asRuntimeException())); + assertFalse(NodeTxExecutor.isRetryable(new InterruptedException("interrupted"))); + // The status is usually wrapped by the time it reaches the retry loop + assertFalse(NodeTxExecutor.isRetryable(HgStoreClientException.of( + "commit failed", new RuntimeException(Status.DEADLINE_EXCEEDED.asRuntimeException())))); + assertTrue(NodeTxExecutor.isRetryable(Status.UNAVAILABLE.asRuntimeException())); + assertTrue(NodeTxExecutor.isRetryable(new RuntimeException("simulated transport failure"))); + } + + @Test + public void testDeadlineExceededIsNotRetried() { + NodeTxExecutor executor = NodeTxExecutor.graphOf("graph", null); + AtomicInteger attempts = new AtomicInteger(); + HgStoreClientException e = assertThrows(HgStoreClientException.class, () -> + executor.retryingInvoke(() -> { + attempts.incrementAndGet(); + throw Status.DEADLINE_EXCEEDED.withDescription("deadline exceeded after 20s") + .asRuntimeException(); + })); + assertEquals(1, attempts.get()); + assertTrue(e.getMessage(), e.getMessage().contains("DEADLINE_EXCEEDED")); + } + + @Test + public void testInterruptStopsRetrying() { + // A REST worker hitting restserver.request_timeout (or a Gremlin evaluationTimeout) + // interrupts the calling thread while the store call is failing; the loop must + // stop instead of sleeping and retrying with the interrupt swallowed. + NodeTxExecutor executor = NodeTxExecutor.graphOf("graph", null); + AtomicInteger attempts = new AtomicInteger(); + try { + assertThrows(HgStoreClientException.class, () -> + executor.retryingInvoke(() -> { + attempts.incrementAndGet(); + Thread.currentThread().interrupt(); + throw new RuntimeException("simulated transport failure"); + })); + assertEquals(1, attempts.get()); + assertTrue("interrupt flag must be restored for the caller", + Thread.currentThread().isInterrupted()); + } finally { + // clear the flag so the test runner thread is not left interrupted + Thread.interrupted(); + } + } + + @Test + public void testTransientFailureIsStillRetried() { + NodeTxExecutor executor = NodeTxExecutor.graphOf("graph", null); + AtomicInteger attempts = new AtomicInteger(); + Optional result = executor.retryingInvoke(() -> { + if (attempts.getAndIncrement() == 0) { + throw Status.UNAVAILABLE.withDescription("store replaced").asRuntimeException(); + } + return "ok"; + }); + assertEquals("ok", result.get()); + assertEquals(2, attempts.get()); + } } From 8adf52217938163bed385ad5b7911f468457238a Mon Sep 17 00:00:00 2001 From: Sebastian Gruza Date: Thu, 10 Sep 2026 23:03:52 +0000 Subject: [PATCH 3/5] fix(store-client): decide retry on every failure of a parallel commit; log the cause; cover the pre-attempt interrupt guard Review follow-ups on #3204: - commitSessions() collects every session failure (ConcurrentLinkedQueue) and throws one HgStoreClientException with the others as suppressed; isRetryable() inspects suppressed failures too (cycle-safe), so a DEADLINE_EXCEEDED on one partition is never retried because an UNAVAILABLE on another happened to be reported first - retryability is checked before the attempt budget, so the last attempt logs the reason that applied - the non-retry warning carries the throwable - tests: mixed-failure commit (one attempt, both sessions rolled back), all-retryable commit, interrupt set before the call skips the attempt; style nits --- .../store/client/NodeTxExecutor.java | 112 +++++++++++------- .../store/client/NodeTxExecutorTest.java | 85 ++++++++++++- 2 files changed, 155 insertions(+), 42 deletions(-) diff --git a/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutor.java b/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutor.java index 4305677302..7c95b3c6d1 100644 --- a/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutor.java +++ b/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutor.java @@ -22,13 +22,18 @@ import static org.apache.hugegraph.store.client.util.HgStoreClientConst.TX_SESSIONS_MAP_CAPACITY; import java.util.Collection; +import java.util.Collections; import java.util.HashMap; +import java.util.IdentityHashMap; +import java.util.Iterator; import java.util.LinkedList; import java.util.List; import java.util.Map; import java.util.Optional; +import java.util.Queue; +import java.util.Set; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicReference; import java.util.function.Function; import java.util.function.Supplier; import java.util.stream.Collector; @@ -131,35 +136,7 @@ void doCommit() { if (!allSuccess.get()) { throw HgStoreClientException.of(msg); } - AtomicReference throwable = new AtomicReference<>(); - Collection sessions = this.sessions.values(); - sessions.parallelStream().forEach(e -> { - if (e.isTx()) { - try { - e.commit(); - } catch (Throwable t) { - throwable.compareAndSet(null, t); - allSuccess.set(false); - } - } - }); - if (!allSuccess.get()) { - if (isTx) { - try { - sessions.stream().forEach(HgStoreSession::rollback); - } catch (Exception e) { - - } - } - Throwable cause = throwable.get(); - if (cause.getCause() != null) { - cause = cause.getCause(); - } - if (cause instanceof HgStoreClientException) { - throw (HgStoreClientException) cause; - } - throw HgStoreClientException.of(cause); - } + this.commitSessions(this.sessions.values()); return true; }); @@ -234,6 +211,52 @@ void doCommit() { // } // }; + /** + * Commit every tx session in parallel. When at least one commit fails, roll back (in tx + * mode) and throw one exception that carries ALL failures: the first one as the cause, the + * others as suppressed. The retry loop looks at all of them, so whether a commit is retried + * no longer depends on which partition happened to fail first. + */ + void commitSessions(Collection sessions) { + Queue failures = new ConcurrentLinkedQueue<>(); + sessions.parallelStream().forEach(e -> { + if (e.isTx()) { + try { + e.commit(); + } catch (Throwable t) { + failures.add(t); + } + } + }); + if (failures.isEmpty()) { + return; + } + if (isTx) { + try { + sessions.stream().forEach(HgStoreSession::rollback); + } catch (Exception e) { + // keep the commit failure as the reported one + } + } + throw aggregate(failures); + } + + static HgStoreClientException aggregate(Collection failures) { + Iterator it = failures.iterator(); + Throwable first = it.next(); + Throwable cause = first.getCause() != null ? first.getCause() : first; + HgStoreClientException result = cause instanceof HgStoreClientException ? + (HgStoreClientException) cause : + HgStoreClientException.of(cause); + while (it.hasNext()) { + Throwable other = it.next(); + if (other != result && other != cause) { + result.addSuppressed(other); + } + } + return result; + } + private boolean doAction(HgTriple nodeParams, Function action) { if (nodeParams.getZ() == null) { @@ -389,17 +412,17 @@ Optional retryingInvoke(Supplier supplier) { try { buffer = supplier.get(); } catch (Throwable t) { - if (i + 1 > NODE_MAX_RETRYING_TIMES) { - log.error(maxTryMsg, t); - throw HgStoreClientException.of( - t.getMessage(), t); - } if (!isRetryable(t)) { // A deadline or a cancellation will not // get better by waiting the full deadline // again; fail fast and let the caller decide. log.warn("Not retrying after: {}", - t.getMessage()); + t.getMessage(), t); + throw HgStoreClientException.of( + t.getMessage(), t); + } + if (i + 1 > NODE_MAX_RETRYING_TIMES) { + log.error(maxTryMsg, t); throw HgStoreClientException.of( t.getMessage(), t); } @@ -428,11 +451,16 @@ Optional retryingInvoke(Supplier supplier) { /** * Retry only failures that a fresh attempt can plausibly fix (a transport error, a * leader change). A DEADLINE_EXCEEDED would wait the full deadline again, a CANCELLED - * means the caller's thread was interrupted; neither is retried. + * means the caller's thread was interrupted; neither is retried, also when it is one of + * several failures of a parallel commit. */ static boolean isRetryable(Throwable t) { + return isRetryable(t, Collections.newSetFromMap(new IdentityHashMap<>())); + } + + private static boolean isRetryable(Throwable t, Set seen) { Throwable c = t; - while (c != null) { + while (c != null && seen.add(c)) { if (c instanceof InterruptedException) { return false; } @@ -442,8 +470,12 @@ static boolean isRetryable(Throwable t) { return false; } } - if (c.getCause() == c) { - break; + // A parallel commit reports the other partitions' failures as suppressed; + // one non-retryable failure among them makes the whole attempt non-retryable. + for (Throwable suppressed : c.getSuppressed()) { + if (!isRetryable(suppressed, seen)) { + return false; + } } c = c.getCause(); } diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutorTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutorTest.java index 629c52c756..c76c963bcc 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutorTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutorTest.java @@ -27,6 +27,8 @@ import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import java.util.Arrays; +import java.util.List; import java.util.Optional; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; @@ -40,7 +42,6 @@ import org.junit.Test; import io.grpc.Status; -import io.grpc.StatusRuntimeException; public class NodeTxExecutorTest { @@ -131,7 +132,13 @@ public void testIsRetryableClassifiesFailures() { assertFalse(NodeTxExecutor.isRetryable(new InterruptedException("interrupted"))); // The status is usually wrapped by the time it reaches the retry loop assertFalse(NodeTxExecutor.isRetryable(HgStoreClientException.of( - "commit failed", new RuntimeException(Status.DEADLINE_EXCEEDED.asRuntimeException())))); + "commit failed", + new RuntimeException(Status.DEADLINE_EXCEEDED.asRuntimeException())))); + // one non-retryable failure among the suppressed ones decides + HgStoreClientException mixed = HgStoreClientException.of( + Status.UNAVAILABLE.asRuntimeException()); + mixed.addSuppressed(Status.DEADLINE_EXCEEDED.asRuntimeException()); + assertFalse(NodeTxExecutor.isRetryable(mixed)); assertTrue(NodeTxExecutor.isRetryable(Status.UNAVAILABLE.asRuntimeException())); assertTrue(NodeTxExecutor.isRetryable(new RuntimeException("simulated transport failure"))); } @@ -186,4 +193,78 @@ public void testTransientFailureIsStillRetried() { assertEquals("ok", result.get()); assertEquals(2, attempts.get()); } + + @Test + public void testInterruptBeforeCallSkipsTheAttempt() { + // restserver.request_timeout expired between two store calls of one request + NodeTxExecutor executor = NodeTxExecutor.graphOf("graph", null); + AtomicInteger attempts = new AtomicInteger(); + try { + Thread.currentThread().interrupt(); + assertThrows(HgStoreClientException.class, () -> + executor.retryingInvoke(() -> { + attempts.incrementAndGet(); + return "ok"; + })); + assertEquals(0, attempts.get()); + assertTrue(Thread.currentThread().isInterrupted()); + } finally { + Thread.interrupted(); + } + } + + @Test + public void testMixedCommitFailuresAreNotRetried() { + // One partition on a store being replaced (UNAVAILABLE, retryable), one on a stalled + // store (DEADLINE_EXCEEDED). Whichever finishes first, the commit must not be retried. + HgStoreSession unavailable = mock(HgStoreSession.class); + HgStoreSession stalled = mock(HgStoreSession.class); + when(unavailable.isTx()).thenReturn(true); + when(stalled.isTx()).thenReturn(true); + // fresh exceptions on every attempt, as the real gRPC calls produce them + org.mockito.Mockito.doAnswer(i -> { + throw new RuntimeException(HgStoreClientException.of( + "commit", Status.UNAVAILABLE.asRuntimeException())); + }).when(unavailable).commit(); + org.mockito.Mockito.doAnswer(i -> { + throw new RuntimeException(HgStoreClientException.of( + "commit", Status.DEADLINE_EXCEEDED.asRuntimeException())); + }).when(stalled).commit(); + List sessions = Arrays.asList(unavailable, stalled); + + NodeTxExecutor executor = NodeTxExecutor.graphOf("graph", null); + executor.setTx(true); + HgStoreClientException e = assertThrows(HgStoreClientException.class, + () -> executor.commitSessions(sessions)); + assertEquals(1, e.getSuppressed().length); + assertFalse(NodeTxExecutor.isRetryable(e)); + verify(unavailable).rollback(); + verify(stalled).rollback(); + + AtomicInteger attempts = new AtomicInteger(); + assertThrows(HgStoreClientException.class, () -> + executor.retryingInvoke(() -> { + attempts.incrementAndGet(); + executor.commitSessions(sessions); + return true; + })); + assertEquals(1, attempts.get()); + } + + @Test + public void testAllRetryableCommitFailuresAreRetried() { + HgStoreSession a = mock(HgStoreSession.class); + HgStoreSession b = mock(HgStoreSession.class); + when(a.isTx()).thenReturn(true); + when(b.isTx()).thenReturn(true); + org.mockito.Mockito.doThrow(new RuntimeException(Status.UNAVAILABLE.asRuntimeException())) + .when(a).commit(); + org.mockito.Mockito.doThrow(new RuntimeException(Status.UNAVAILABLE.asRuntimeException())) + .when(b).commit(); + NodeTxExecutor executor = NodeTxExecutor.graphOf("graph", null); + HgStoreClientException e = assertThrows(HgStoreClientException.class, + () -> executor.commitSessions(Arrays.asList(a, b))); + assertEquals(1, e.getSuppressed().length); + assertTrue(NodeTxExecutor.isRetryable(e)); + } } From b588aa8882c101f26114484134a23e89f4b2ab7d Mon Sep 17 00:00:00 2001 From: Sebastian Gruza Date: Fri, 11 Sep 2026 10:09:36 +0000 Subject: [PATCH 4/5] fix(store-client): retry DEADLINE_EXCEEDED exactly once so a moved partition leader is still reached Review follow-up on #3204: the NOT_WORK notice sent for the failed RPC reloads the partition leaders, so with replicated partitions the next attempt can reach a new raft leader. retryingInvoke() now classifies a failure as FATAL (interrupt, CANCELLED: never retried), DEADLINE (retried once per call; a second deadline in a row would only wait the full deadline again on the same stalled store) or RETRYABLE (as before). Suppressed failures of a parallel commit are classified too, the most severe class wins. Tests: classification, deadline retried exactly once, deadline then new leader succeeds, mixed commit retried once. --- .../store/client/NodeTxExecutor.java | 70 ++++++++++++++----- .../store/client/NodeTxExecutorTest.java | 59 +++++++++++----- 2 files changed, 91 insertions(+), 38 deletions(-) diff --git a/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutor.java b/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutor.java index 7c95b3c6d1..5a26b85214 100644 --- a/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutor.java +++ b/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutor.java @@ -398,6 +398,7 @@ boolean ifAnyTrue(Supplier>> nodeStreamSuppl } Optional retryingInvoke(Supplier supplier) { + boolean[] deadlineRetried = {false}; return IntStream.rangeClosed(0, NODE_MAX_RETRYING_TIMES).boxed() .map( i -> { @@ -412,15 +413,32 @@ Optional retryingInvoke(Supplier supplier) { try { buffer = supplier.get(); } catch (Throwable t) { - if (!isRetryable(t)) { - // A deadline or a cancellation will not - // get better by waiting the full deadline - // again; fail fast and let the caller decide. + Failure failure = classify(t); + if (failure == Failure.FATAL) { + // The caller's thread was interrupted or the + // call was cancelled: fail fast. log.warn("Not retrying after: {}", t.getMessage(), t); throw HgStoreClientException.of( t.getMessage(), t); } + if (failure == Failure.DEADLINE) { + // One retry: the NOT_WORK notice sent for the + // failed RPC reloads the partition leaders, so + // the next attempt can reach a new leader. A + // second deadline in a row would only wait the + // full deadline again on the same stalled store. + if (deadlineRetried[0]) { + log.warn("Not retrying a second deadline: {}", + t.getMessage(), t); + throw HgStoreClientException.of( + t.getMessage(), t); + } + deadlineRetried[0] = true; + log.warn("Deadline exceeded, retrying once in " + + "case the partition leader moved: {}", + t.getMessage()); + } if (i + 1 > NODE_MAX_RETRYING_TIMES) { log.error(maxTryMsg, t); throw HgStoreClientException.of( @@ -449,37 +467,51 @@ Optional retryingInvoke(Supplier supplier) { } /** - * Retry only failures that a fresh attempt can plausibly fix (a transport error, a - * leader change). A DEADLINE_EXCEEDED would wait the full deadline again, a CANCELLED - * means the caller's thread was interrupted; neither is retried, also when it is one of - * several failures of a parallel commit. + * How a failed attempt is treated by {@link #retryingInvoke}: FATAL is never retried (the + * caller's thread was interrupted, or the call was cancelled), DEADLINE is retried exactly + * once (the partition leader may have moved after the failed RPC invalidated the partition + * cache; a second deadline in a row would only wait the full deadline again on the same + * stalled store), everything else (transport errors, NOT_LEADER, store replacement) is + * retried up to NODE_MAX_RETRYING_TIMES as before. For a parallel commit the failures of + * the other partitions arrive as suppressed exceptions and are classified too; the most + * severe class wins. */ - static boolean isRetryable(Throwable t) { - return isRetryable(t, Collections.newSetFromMap(new IdentityHashMap<>())); + enum Failure { + RETRYABLE, DEADLINE, FATAL + } + + static Failure classify(Throwable t) { + return classify(t, Collections.newSetFromMap(new IdentityHashMap<>())); } - private static boolean isRetryable(Throwable t, Set seen) { + private static Failure classify(Throwable t, Set seen) { + Failure worst = Failure.RETRYABLE; Throwable c = t; while (c != null && seen.add(c)) { if (c instanceof InterruptedException) { - return false; + return Failure.FATAL; } if (c instanceof StatusRuntimeException) { Status.Code code = ((StatusRuntimeException) c).getStatus().getCode(); - if (code == Status.Code.DEADLINE_EXCEEDED || code == Status.Code.CANCELLED) { - return false; + if (code == Status.Code.CANCELLED) { + return Failure.FATAL; + } + if (code == Status.Code.DEADLINE_EXCEEDED) { + worst = Failure.DEADLINE; } } - // A parallel commit reports the other partitions' failures as suppressed; - // one non-retryable failure among them makes the whole attempt non-retryable. for (Throwable suppressed : c.getSuppressed()) { - if (!isRetryable(suppressed, seen)) { - return false; + Failure f = classify(suppressed, seen); + if (f == Failure.FATAL) { + return f; + } + if (f.compareTo(worst) > 0) { + worst = f; } } c = c.getCause(); } - return true; + return worst; } private boolean isValid(Object obj) { diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutorTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutorTest.java index c76c963bcc..534ee50f06 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutorTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutorTest.java @@ -18,7 +18,6 @@ package org.apache.hugegraph.store.client; import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertSame; import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; @@ -126,25 +125,32 @@ public void testParallelReplacementUsesOneCurrentSession() throws Exception { } @Test - public void testIsRetryableClassifiesFailures() { - assertFalse(NodeTxExecutor.isRetryable(Status.DEADLINE_EXCEEDED.asRuntimeException())); - assertFalse(NodeTxExecutor.isRetryable(Status.CANCELLED.asRuntimeException())); - assertFalse(NodeTxExecutor.isRetryable(new InterruptedException("interrupted"))); + public void testClassifyFailures() { + assertEquals(NodeTxExecutor.Failure.DEADLINE, + NodeTxExecutor.classify(Status.DEADLINE_EXCEEDED.asRuntimeException())); + assertEquals(NodeTxExecutor.Failure.FATAL, + NodeTxExecutor.classify(Status.CANCELLED.asRuntimeException())); + assertEquals(NodeTxExecutor.Failure.FATAL, + NodeTxExecutor.classify(new InterruptedException("interrupted"))); // The status is usually wrapped by the time it reaches the retry loop - assertFalse(NodeTxExecutor.isRetryable(HgStoreClientException.of( - "commit failed", - new RuntimeException(Status.DEADLINE_EXCEEDED.asRuntimeException())))); - // one non-retryable failure among the suppressed ones decides + assertEquals(NodeTxExecutor.Failure.DEADLINE, NodeTxExecutor.classify( + HgStoreClientException.of("commit", new RuntimeException( + Status.DEADLINE_EXCEEDED.asRuntimeException())))); + // The most severe class among the suppressed failures of a parallel commit wins HgStoreClientException mixed = HgStoreClientException.of( Status.UNAVAILABLE.asRuntimeException()); mixed.addSuppressed(Status.DEADLINE_EXCEEDED.asRuntimeException()); - assertFalse(NodeTxExecutor.isRetryable(mixed)); - assertTrue(NodeTxExecutor.isRetryable(Status.UNAVAILABLE.asRuntimeException())); - assertTrue(NodeTxExecutor.isRetryable(new RuntimeException("simulated transport failure"))); + assertEquals(NodeTxExecutor.Failure.DEADLINE, NodeTxExecutor.classify(mixed)); + mixed.addSuppressed(Status.CANCELLED.asRuntimeException()); + assertEquals(NodeTxExecutor.Failure.FATAL, NodeTxExecutor.classify(mixed)); + assertEquals(NodeTxExecutor.Failure.RETRYABLE, + NodeTxExecutor.classify(Status.UNAVAILABLE.asRuntimeException())); + assertEquals(NodeTxExecutor.Failure.RETRYABLE, + NodeTxExecutor.classify(new RuntimeException("simulated transport failure"))); } @Test - public void testDeadlineExceededIsNotRetried() { + public void testDeadlineExceededIsRetriedExactlyOnce() { NodeTxExecutor executor = NodeTxExecutor.graphOf("graph", null); AtomicInteger attempts = new AtomicInteger(); HgStoreClientException e = assertThrows(HgStoreClientException.class, () -> @@ -153,10 +159,25 @@ public void testDeadlineExceededIsNotRetried() { throw Status.DEADLINE_EXCEEDED.withDescription("deadline exceeded after 20s") .asRuntimeException(); })); - assertEquals(1, attempts.get()); + assertEquals(2, attempts.get()); assertTrue(e.getMessage(), e.getMessage().contains("DEADLINE_EXCEEDED")); } + @Test + public void testDeadlineThenNewLeaderSucceeds() { + // the failed RPC invalidates the partition cache; the single retry reaches the new leader + NodeTxExecutor executor = NodeTxExecutor.graphOf("graph", null); + AtomicInteger attempts = new AtomicInteger(); + Optional result = executor.retryingInvoke(() -> { + if (attempts.getAndIncrement() == 0) { + throw Status.DEADLINE_EXCEEDED.asRuntimeException(); + } + return "ok"; + }); + assertEquals("ok", result.get()); + assertEquals(2, attempts.get()); + } + @Test public void testInterruptStopsRetrying() { // A REST worker hitting restserver.request_timeout (or a Gremlin evaluationTimeout) @@ -214,9 +235,9 @@ public void testInterruptBeforeCallSkipsTheAttempt() { } @Test - public void testMixedCommitFailuresAreNotRetried() { + public void testMixedCommitFailuresAreRetriedExactlyOnce() { // One partition on a store being replaced (UNAVAILABLE, retryable), one on a stalled - // store (DEADLINE_EXCEEDED). Whichever finishes first, the commit must not be retried. + // store (DEADLINE_EXCEEDED). Whichever finishes first, the deadline decides: one retry. HgStoreSession unavailable = mock(HgStoreSession.class); HgStoreSession stalled = mock(HgStoreSession.class); when(unavailable.isTx()).thenReturn(true); @@ -237,7 +258,7 @@ public void testMixedCommitFailuresAreNotRetried() { HgStoreClientException e = assertThrows(HgStoreClientException.class, () -> executor.commitSessions(sessions)); assertEquals(1, e.getSuppressed().length); - assertFalse(NodeTxExecutor.isRetryable(e)); + assertEquals(NodeTxExecutor.Failure.DEADLINE, NodeTxExecutor.classify(e)); verify(unavailable).rollback(); verify(stalled).rollback(); @@ -248,7 +269,7 @@ public void testMixedCommitFailuresAreNotRetried() { executor.commitSessions(sessions); return true; })); - assertEquals(1, attempts.get()); + assertEquals(2, attempts.get()); } @Test @@ -265,6 +286,6 @@ public void testAllRetryableCommitFailuresAreRetried() { HgStoreClientException e = assertThrows(HgStoreClientException.class, () -> executor.commitSessions(Arrays.asList(a, b))); assertEquals(1, e.getSuppressed().length); - assertTrue(NodeTxExecutor.isRetryable(e)); + assertEquals(NodeTxExecutor.Failure.RETRYABLE, NodeTxExecutor.classify(e)); } } From 172b2d03c3e07e4857d982e6ab978f1e1f6b8160 Mon Sep 17 00:00:00 2001 From: Sebastian Gruza Date: Fri, 11 Sep 2026 16:58:53 +0000 Subject: [PATCH 5/5] fix(store-client): keep InterruptedException as the root cause of the interrupt exits Review follow-up on #3204: HugeException.isInterrupted() checks the root cause and HugeTask.fail() relies on it to record a cancelled task as CANCELLED rather than FAILED. Both new interrupt exits of retryingInvoke() now carry an InterruptedException as the root cause (the store failure that was being retried is attached as suppressed). Tests assert the root cause on both exits. --- .../hugegraph/store/client/NodeTxExecutor.java | 9 +++++++-- .../store/client/NodeTxExecutorTest.java | 16 ++++++++++++++-- 2 files changed, 21 insertions(+), 4 deletions(-) diff --git a/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutor.java b/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutor.java index 5a26b85214..f0c647f5b9 100644 --- a/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutor.java +++ b/hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutor.java @@ -406,8 +406,12 @@ Optional retryingInvoke(Supplier supplier) { // The caller (e.g. a REST worker hitting // restserver.request_timeout) gave up: stop // retrying instead of holding its thread. + // InterruptedException as the root cause: the + // server's task cancel path recognises it + // (HugeException.isInterrupted()). throw HgStoreClientException.of( - "Interrupted before retry " + i); + "Interrupted before retry " + i, + new InterruptedException()); } T buffer = null; try { @@ -453,9 +457,10 @@ Optional retryingInvoke(Supplier supplier) { Thread.sleep(sleepTime * 1000L); } catch (InterruptedException e) { Thread.currentThread().interrupt(); + e.addSuppressed(t); throw HgStoreClientException.of( "Interrupted while waiting to retry: " + - t.getMessage(), t); + t.getMessage(), e); } } return buffer; diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutorTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutorTest.java index 534ee50f06..d39b079c65 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutorTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutorTest.java @@ -186,7 +186,7 @@ public void testInterruptStopsRetrying() { NodeTxExecutor executor = NodeTxExecutor.graphOf("graph", null); AtomicInteger attempts = new AtomicInteger(); try { - assertThrows(HgStoreClientException.class, () -> + HgStoreClientException e = assertThrows(HgStoreClientException.class, () -> executor.retryingInvoke(() -> { attempts.incrementAndGet(); Thread.currentThread().interrupt(); @@ -195,12 +195,23 @@ public void testInterruptStopsRetrying() { assertEquals(1, attempts.get()); assertTrue("interrupt flag must be restored for the caller", Thread.currentThread().isInterrupted()); + // HugeException.isInterrupted() looks at the root cause + assertTrue(rootCause(e) instanceof InterruptedException); + assertEquals("simulated transport failure", + rootCause(e).getSuppressed()[0].getMessage()); } finally { // clear the flag so the test runner thread is not left interrupted Thread.interrupted(); } } + private static Throwable rootCause(Throwable t) { + while (t.getCause() != null && t.getCause() != t) { + t = t.getCause(); + } + return t; + } + @Test public void testTransientFailureIsStillRetried() { NodeTxExecutor executor = NodeTxExecutor.graphOf("graph", null); @@ -222,13 +233,14 @@ public void testInterruptBeforeCallSkipsTheAttempt() { AtomicInteger attempts = new AtomicInteger(); try { Thread.currentThread().interrupt(); - assertThrows(HgStoreClientException.class, () -> + HgStoreClientException e = assertThrows(HgStoreClientException.class, () -> executor.retryingInvoke(() -> { attempts.incrementAndGet(); return "ok"; })); assertEquals(0, attempts.get()); assertTrue(Thread.currentThread().isInterrupted()); + assertTrue(rootCause(e) instanceof InterruptedException); } finally { Thread.interrupted(); }