@Test public void retryReachesMaximumAndForwardException() throws Throwable { RuntimeException oops = new RuntimeException("oops"); when(joinPoint.proceed()).thenThrow(oops); when(compensable.retries()).thenReturn(3); try { aspect.advise(joinPoint, compensable); expectFailing(RuntimeException.class); } catch (RuntimeException e) { assertThat(e.getMessage(), is("oops")); } assertThat(messages.size(), is(6)); TxEvent startedEvent1 = messages.get(0); assertThat(startedEvent1.globalTxId(), is(globalTxId)); assertThat(startedEvent1.localTxId(), is(newLocalTxId)); assertThat(startedEvent1.parentTxId(), is(localTxId)); assertThat(startedEvent1.type(), is(EventType.TxStartedEvent)); assertThat(startedEvent1.retries(), is(3)); assertThat(startedEvent1.retryMethod(), is(this.getClass().getDeclaredMethod("doNothing").toString())); assertThat(messages.get(1).type(), is(EventType.TxAbortedEvent)); TxEvent startedEvent2 = messages.get(2); assertThat(startedEvent2.localTxId(), is(newLocalTxId)); assertThat(startedEvent2.type(), is(EventType.TxStartedEvent)); assertThat(startedEvent2.retries(), is(2)); assertThat(messages.get(3).type(), is(EventType.TxAbortedEvent)); TxEvent startedEvent3 = messages.get(4); assertThat(startedEvent3.localTxId(), is(newLocalTxId)); assertThat(startedEvent3.type(), is(EventType.TxStartedEvent)); assertThat(startedEvent3.retries(), is(1)); assertThat(messages.get(5).type(), is(EventType.TxAbortedEvent)); assertThat(omegaContext.globalTxId(), is(globalTxId)); assertThat(omegaContext.localTxId(), is(localTxId)); }
@Around("execution(@org.apache.servicecomb.saga.omega.transaction.annotations.Compensable * *(..)) && @annotation(compensable)") Object advise(ProceedingJoinPoint joinPoint, Compensable compensable) throws Throwable { Method method = ((MethodSignature) joinPoint.getSignature()).getMethod(); String localTxId = context.localTxId(); context.newLocalTxId(); LOG.debug("Updated context {} for compensable method {} ", context, method.toString()); int retries = compensable.retries(); RecoveryPolicy recoveryPolicy = RecoveryPolicyFactory.getRecoveryPolicy(retries); try { return recoveryPolicy.apply(joinPoint, compensable, interceptor, context, localTxId, retries); } catch (Exception e) { throw e; } finally { context.setLocalTxId(localTxId); LOG.debug("Restored context back to {}", context); } }
TransactionAspect { @Around("execution(@org.apache.servicecomb.saga.omega.transaction.annotations.Compensable * *(..)) && @annotation(compensable)") Object advise(ProceedingJoinPoint joinPoint, Compensable compensable) throws Throwable { Method method = ((MethodSignature) joinPoint.getSignature()).getMethod(); String localTxId = context.localTxId(); context.newLocalTxId(); LOG.debug("Updated context {} for compensable method {} ", context, method.toString()); int retries = compensable.retries(); RecoveryPolicy recoveryPolicy = RecoveryPolicyFactory.getRecoveryPolicy(retries); try { return recoveryPolicy.apply(joinPoint, compensable, interceptor, context, localTxId, retries); } catch (Exception e) { throw e; } finally { context.setLocalTxId(localTxId); LOG.debug("Restored context back to {}", context); } } }
TransactionAspect { @Around("execution(@org.apache.servicecomb.saga.omega.transaction.annotations.Compensable * *(..)) && @annotation(compensable)") Object advise(ProceedingJoinPoint joinPoint, Compensable compensable) throws Throwable { Method method = ((MethodSignature) joinPoint.getSignature()).getMethod(); String localTxId = context.localTxId(); context.newLocalTxId(); LOG.debug("Updated context {} for compensable method {} ", context, method.toString()); int retries = compensable.retries(); RecoveryPolicy recoveryPolicy = RecoveryPolicyFactory.getRecoveryPolicy(retries); try { return recoveryPolicy.apply(joinPoint, compensable, interceptor, context, localTxId, retries); } catch (Exception e) { throw e; } finally { context.setLocalTxId(localTxId); LOG.debug("Restored context back to {}", context); } } TransactionAspect(MessageSender sender, OmegaContext context); }
TransactionAspect { @Around("execution(@org.apache.servicecomb.saga.omega.transaction.annotations.Compensable * *(..)) && @annotation(compensable)") Object advise(ProceedingJoinPoint joinPoint, Compensable compensable) throws Throwable { Method method = ((MethodSignature) joinPoint.getSignature()).getMethod(); String localTxId = context.localTxId(); context.newLocalTxId(); LOG.debug("Updated context {} for compensable method {} ", context, method.toString()); int retries = compensable.retries(); RecoveryPolicy recoveryPolicy = RecoveryPolicyFactory.getRecoveryPolicy(retries); try { return recoveryPolicy.apply(joinPoint, compensable, interceptor, context, localTxId, retries); } catch (Exception e) { throw e; } finally { context.setLocalTxId(localTxId); LOG.debug("Restored context back to {}", context); } } TransactionAspect(MessageSender sender, OmegaContext context);  }
TransactionAspect { @Around("execution(@org.apache.servicecomb.saga.omega.transaction.annotations.Compensable * *(..)) && @annotation(compensable)") Object advise(ProceedingJoinPoint joinPoint, Compensable compensable) throws Throwable { Method method = ((MethodSignature) joinPoint.getSignature()).getMethod(); String localTxId = context.localTxId(); context.newLocalTxId(); LOG.debug("Updated context {} for compensable method {} ", context, method.toString()); int retries = compensable.retries(); RecoveryPolicy recoveryPolicy = RecoveryPolicyFactory.getRecoveryPolicy(retries); try { return recoveryPolicy.apply(joinPoint, compensable, interceptor, context, localTxId, retries); } catch (Exception e) { throw e; } finally { context.setLocalTxId(localTxId); LOG.debug("Restored context back to {}", context); } } TransactionAspect(MessageSender sender, OmegaContext context);   }