### Question:
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);  }### Answer:
@Test public void forwardExceptionWhenGlobalTxAborted() { MessageSender sender = mock(MessageSender.class); when(sender.send(any(TxEvent.class))).thenReturn(new AlphaResponse(true)); CompensableInterceptor interceptor = new CompensableInterceptor(omegaContext, sender); try { recoveryPolicy.apply(joinPoint, compensable, interceptor, omegaContext, parentTxId, 0); expectFailing(InvalidTransactionException.class); } catch (InvalidTransactionException e) { assertThat(e.getMessage().contains("Abort sub transaction"), is(true)); } catch (Throwable throwable) { fail("unexpected exception throw: " + throwable); } verify(sender, times(1)).send(any(TxEvent.class)); }
@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)); }
@Test public void keepRetryingTillInterrupted() throws Throwable { when(compensable.retries()).thenReturn(-1); when(compensable.retryDelayInMilliseconds()).thenReturn(1000); when(joinPoint.proceed()).thenThrow(oops); Thread thread = new Thread(new Runnable() { @Override public void run() { try { recoveryPolicy.apply(joinPoint, compensable, interceptor, omegaContext, parentTxId, -1); expectFailing(OmegaException.class); } catch (OmegaException e) { exception = e; } catch (Throwable throwable) { fail("unexpected exception throw: " + throwable); } } }); thread.start(); thread.interrupt(); thread.join(); assertThat(exception.getMessage().contains("Failed to handle tx because it is interrupted"), is(true)); }