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..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 @@ -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; @@ -46,6 +51,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; /** @@ -129,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; }); @@ -232,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) { @@ -373,35 +398,70 @@ boolean ifAnyTrue(Supplier>> nodeStreamSuppl } Optional retryingInvoke(Supplier supplier) { + boolean[] deadlineRetried = {false}; 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. + // InterruptedException as the root cause: the + // server's task cancel path recognises it + // (HugeException.isInterrupted()). + throw HgStoreClientException.of( + "Interrupted before retry " + i, + new InterruptedException()); + } 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); + 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); } - } else { + 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( 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(); + e.addSuppressed(t); + throw HgStoreClientException.of( + "Interrupted while waiting to retry: " + + t.getMessage(), e); + } } return buffer; } @@ -411,6 +471,54 @@ Optional retryingInvoke(Supplier supplier) { } + /** + * 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. + */ + enum Failure { + RETRYABLE, DEADLINE, FATAL + } + + static Failure classify(Throwable t) { + return classify(t, Collections.newSetFromMap(new IdentityHashMap<>())); + } + + 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 Failure.FATAL; + } + if (c instanceof StatusRuntimeException) { + Status.Code code = ((StatusRuntimeException) c).getStatus().getCode(); + if (code == Status.Code.CANCELLED) { + return Failure.FATAL; + } + if (code == Status.Code.DEADLINE_EXCEEDED) { + worst = Failure.DEADLINE; + } + } + for (Throwable suppressed : c.getSuppressed()) { + Failure f = classify(suppressed, seen); + if (f == Failure.FATAL) { + return f; + } + if (f.compareTo(worst) > 0) { + worst = f; + } + } + c = c.getCause(); + } + return worst; + } + private boolean isValid(Object obj) { if (obj == null) { return false; 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..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 @@ -17,12 +17,17 @@ package org.apache.hugegraph.store.client; +import static org.junit.Assert.assertEquals; 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; 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; @@ -32,8 +37,11 @@ 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; + public class NodeTxExecutorTest { @Test @@ -115,4 +123,181 @@ public void testParallelReplacementUsesOneCurrentSession() throws Exception { workers.shutdownNow(); } } + + @Test + 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 + 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()); + 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 testDeadlineExceededIsRetriedExactlyOnce() { + 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(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) + // 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 { + HgStoreClientException e = 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()); + // 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); + 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()); + } + + @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(); + 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(); + } + } + + @Test + public void testMixedCommitFailuresAreRetriedExactlyOnce() { + // One partition on a store being replaced (UNAVAILABLE, retryable), one on a stalled + // 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); + 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); + assertEquals(NodeTxExecutor.Failure.DEADLINE, NodeTxExecutor.classify(e)); + verify(unavailable).rollback(); + verify(stalled).rollback(); + + AtomicInteger attempts = new AtomicInteger(); + assertThrows(HgStoreClientException.class, () -> + executor.retryingInvoke(() -> { + attempts.incrementAndGet(); + executor.commitSessions(sessions); + return true; + })); + assertEquals(2, 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); + assertEquals(NodeTxExecutor.Failure.RETRYABLE, NodeTxExecutor.classify(e)); + } }