@Test public void throwExceptionWhenRetryReachesMaximum() throws Throwable { when(compensable.retries()).thenReturn(2); when(joinPoint.proceed()).thenThrow(oops); try { recoveryPolicy.apply(joinPoint, compensable, interceptor, omegaContext, parentTxId, 2); expectFailing(RuntimeException.class); } catch (RuntimeException e) { assertThat(e.getMessage(), is("oops")); } assertThat(messages.size(), is(4)); assertThat(messages.get(0).type(), is(EventType.TxStartedEvent)); assertThat(messages.get(1).type(), is(EventType.TxAbortedEvent)); assertThat(messages.get(2).type(), is(EventType.TxStartedEvent)); assertThat(messages.get(3).type(), is(EventType.TxAbortedEvent)); }
@Override public Object apply(ProceedingJoinPoint joinPoint, Compensable compensable, CompensableInterceptor interceptor, OmegaContext context, String parentTxId, int retries) throws Throwable { Method method = ((MethodSignature) joinPoint.getSignature()).getMethod(); int remains = retries; try { while (true) { try { return super.apply(joinPoint, compensable, interceptor, context, parentTxId, remains); } catch (Throwable throwable) { if (throwable instanceof InvalidTransactionException) { throw throwable; } if (remains > 0) { remains--; } else if (remains == 0) { LOG.error( "Retried sub tx failed maximum times, global tx id: {}, local tx id: {}, method: {}, retried times: {}", context.globalTxId(), context.localTxId(), method.toString(), retries); throw throwable; } LOG.warn("Retrying sub tx failed, global tx id: {}, local tx id: {}, method: {}, remains: {}", context.globalTxId(), context.localTxId(), method.toString(), remains); Thread.sleep(compensable.retryDelayInMilliseconds()); } } } catch (InterruptedException e) { String errorMessage = "Failed to handle tx because it is interrupted, global tx id: " + context.globalTxId() + ", local tx id: " + context.localTxId() + ", method: " + method.toString(); LOG.error(errorMessage); interceptor.onError(parentTxId, compensationMethodSignature(joinPoint, compensable, method), e); throw new OmegaException(errorMessage); } catch (Exception e) { throw e; } }
ForwardRecovery extends DefaultRecovery { @Override public Object apply(ProceedingJoinPoint joinPoint, Compensable compensable, CompensableInterceptor interceptor, OmegaContext context, String parentTxId, int retries) throws Throwable { Method method = ((MethodSignature) joinPoint.getSignature()).getMethod(); int remains = retries; try { while (true) { try { return super.apply(joinPoint, compensable, interceptor, context, parentTxId, remains); } catch (Throwable throwable) { if (throwable instanceof InvalidTransactionException) { throw throwable; } if (remains > 0) { remains--; } else if (remains == 0) { LOG.error( "Retried sub tx failed maximum times, global tx id: {}, local tx id: {}, method: {}, retried times: {}", context.globalTxId(), context.localTxId(), method.toString(), retries); throw throwable; } LOG.warn("Retrying sub tx failed, global tx id: {}, local tx id: {}, method: {}, remains: {}", context.globalTxId(), context.localTxId(), method.toString(), remains); Thread.sleep(compensable.retryDelayInMilliseconds()); } } } catch (InterruptedException e) { String errorMessage = "Failed to handle tx because it is interrupted, global tx id: " + context.globalTxId() + ", local tx id: " + context.localTxId() + ", method: " + method.toString(); LOG.error(errorMessage); interceptor.onError(parentTxId, compensationMethodSignature(joinPoint, compensable, method), e); throw new OmegaException(errorMessage); } catch (Exception e) { throw e; } } }
ForwardRecovery extends DefaultRecovery { @Override public Object apply(ProceedingJoinPoint joinPoint, Compensable compensable, CompensableInterceptor interceptor, OmegaContext context, String parentTxId, int retries) throws Throwable { Method method = ((MethodSignature) joinPoint.getSignature()).getMethod(); int remains = retries; try { while (true) { try { return super.apply(joinPoint, compensable, interceptor, context, parentTxId, remains); } catch (Throwable throwable) { if (throwable instanceof InvalidTransactionException) { throw throwable; } if (remains > 0) { remains--; } else if (remains == 0) { LOG.error( "Retried sub tx failed maximum times, global tx id: {}, local tx id: {}, method: {}, retried times: {}", context.globalTxId(), context.localTxId(), method.toString(), retries); throw throwable; } LOG.warn("Retrying sub tx failed, global tx id: {}, local tx id: {}, method: {}, remains: {}", context.globalTxId(), context.localTxId(), method.toString(), remains); Thread.sleep(compensable.retryDelayInMilliseconds()); } } } catch (InterruptedException e) { String errorMessage = "Failed to handle tx because it is interrupted, global tx id: " + context.globalTxId() + ", local tx id: " + context.localTxId() + ", method: " + method.toString(); LOG.error(errorMessage); interceptor.onError(parentTxId, compensationMethodSignature(joinPoint, compensable, method), e); throw new OmegaException(errorMessage); } catch (Exception e) { throw e; } }  }
ForwardRecovery extends DefaultRecovery { @Override public Object apply(ProceedingJoinPoint joinPoint, Compensable compensable, CompensableInterceptor interceptor, OmegaContext context, String parentTxId, int retries) throws Throwable { Method method = ((MethodSignature) joinPoint.getSignature()).getMethod(); int remains = retries; try { while (true) { try { return super.apply(joinPoint, compensable, interceptor, context, parentTxId, remains); } catch (Throwable throwable) { if (throwable instanceof InvalidTransactionException) { throw throwable; } if (remains > 0) { remains--; } else if (remains == 0) { LOG.error( "Retried sub tx failed maximum times, global tx id: {}, local tx id: {}, method: {}, retried times: {}", context.globalTxId(), context.localTxId(), method.toString(), retries); throw throwable; } LOG.warn("Retrying sub tx failed, global tx id: {}, local tx id: {}, method: {}, remains: {}", context.globalTxId(), context.localTxId(), method.toString(), remains); Thread.sleep(compensable.retryDelayInMilliseconds()); } } } catch (InterruptedException e) { String errorMessage = "Failed to handle tx because it is interrupted, global tx id: " + context.globalTxId() + ", local tx id: " + context.localTxId() + ", method: " + method.toString(); LOG.error(errorMessage); interceptor.onError(parentTxId, compensationMethodSignature(joinPoint, compensable, method), e); throw new OmegaException(errorMessage); } catch (Exception e) { throw e; } }  @Override Object apply(ProceedingJoinPoint joinPoint, Compensable compensable, CompensableInterceptor interceptor,
      OmegaContext context, String parentTxId, int retries); }
ForwardRecovery extends DefaultRecovery { @Override public Object apply(ProceedingJoinPoint joinPoint, Compensable compensable, CompensableInterceptor interceptor, OmegaContext context, String parentTxId, int retries) throws Throwable { Method method = ((MethodSignature) joinPoint.getSignature()).getMethod(); int remains = retries; try { while (true) { try { return super.apply(joinPoint, compensable, interceptor, context, parentTxId, remains); } catch (Throwable throwable) { if (throwable instanceof InvalidTransactionException) { throw throwable; } if (remains > 0) { remains--; } else if (remains == 0) { LOG.error( "Retried sub tx failed maximum times, global tx id: {}, local tx id: {}, method: {}, retried times: {}", context.globalTxId(), context.localTxId(), method.toString(), retries); throw throwable; } LOG.warn("Retrying sub tx failed, global tx id: {}, local tx id: {}, method: {}, remains: {}", context.globalTxId(), context.localTxId(), method.toString(), remains); Thread.sleep(compensable.retryDelayInMilliseconds()); } } } catch (InterruptedException e) { String errorMessage = "Failed to handle tx because it is interrupted, global tx id: " + context.globalTxId() + ", local tx id: " + context.localTxId() + ", method: " + method.toString(); LOG.error(errorMessage); interceptor.onError(parentTxId, compensationMethodSignature(joinPoint, compensable, method), e); throw new OmegaException(errorMessage); } catch (Exception e) { throw e; } }  @Override Object apply(ProceedingJoinPoint joinPoint, Compensable compensable, CompensableInterceptor interceptor,
      OmegaContext context, String parentTxId, int retries);  }