SebastianGruza commented on code in PR #3204:
URL: https://github.com/apache/hugegraph/pull/3204#discussion_r3995591672
##########
hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutor.java:
##########
@@ -129,35 +136,7 @@ void doCommit() {
if (!allSuccess.get()) {
throw HgStoreClientException.of(msg);
}
- AtomicReference<Throwable> throwable = new AtomicReference<>();
- Collection<HgStoreSession> 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());
Review Comment:
Done in 35e0a6d — `allSuccess`, the unreachable throw and the unused `msg`
constant are gone (the `doAction()` return value was never wired into that gate
before either, so nothing changes behaviourally).
##########
hugegraph-store/hg-store-client/src/main/java/org/apache/hugegraph/store/client/NodeTxExecutor.java:
##########
@@ -373,35 +398,70 @@ boolean ifAnyTrue(Supplier<Stream<HgPair<HgStoreNode,
NodeTkv>>> nodeStreamSuppl
}
<T> Optional<T> retryingInvoke(Supplier<T> 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]) {
Review Comment:
Done in 35e0a6d — I kept the "in a row" semantics and made the code match:
the budget flag is reset whenever an attempt fails with a non-DEADLINE
classification, so deadline → UNAVAILABLE → deadline retries the second
deadline too (`testDeadlineBudgetResetsAfterAnotherFailure`, 4 attempts). 11/11
locally.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
