@Test public void returnImmediatelyWhenReceivedRejectResponse() { 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)); }
@Override public Object apply(ProceedingJoinPoint joinPoint, Compensable compensable, CompensableInterceptor interceptor, OmegaContext context, String parentTxId, int retries) throws Throwable { Method method = ((MethodSignature) joinPoint.getSignature()).getMethod(); LOG.debug("Intercepting compensable method {} with context {}", method.toString(), context); String compensationSignature = compensable.compensationMethod().isEmpty() ? "" : compensationMethodSignature(joinPoint, compensable, method); String retrySignature = (retries != 0 || compensationSignature.isEmpty()) ? method.toString() : ""; boolean isProceed = false; boolean enabledTx = false; try { CurrentThreadOmegaContext.putThreadGlobalLocalTxId(new OmegaContextServiceConfig(context, false, false)); AlphaResponse response = interceptor.preIntercept(parentTxId, compensationSignature, compensable.timeout(), retrySignature, retries, joinPoint.getArgs()); enabledTx = response.enabledTx(); isProceed = true; Object result = null; if (!response.aborted()) { result = joinPoint.proceed(); } if (enabledTx) { if (response.aborted()) { String abortedLocalTxId = context.localTxId(); context.setLocalTxId(parentTxId); throw new InvalidTransactionException("Abort sub transaction " + abortedLocalTxId + " because global transaction " + context.globalTxId() + " has already aborted."); } CurrentThreadOmegaContext.clearCache(); interceptor.postIntercept(parentTxId, compensationSignature); } return result; } catch (InvalidTransactionException ite) { throw ite; } catch (Throwable throwable) { boolean isFaultTolerant = ApplicationContextUtil.getApplicationContext().getBean(MessageSender.class).readConfigFromServer(ConfigCenterType.CompensationFaultTolerant.toInteger(), context.category()).getStatus(); if (enabledTx && !isFaultTolerant) { interceptor.onError(parentTxId, compensationSignature, throwable); } if (!isProceed && isFaultTolerant) { return joinPoint.proceed(); } throw throwable; } }
DefaultRecovery implements RecoveryPolicy { @Override public Object apply(ProceedingJoinPoint joinPoint, Compensable compensable, CompensableInterceptor interceptor, OmegaContext context, String parentTxId, int retries) throws Throwable { Method method = ((MethodSignature) joinPoint.getSignature()).getMethod(); LOG.debug("Intercepting compensable method {} with context {}", method.toString(), context); String compensationSignature = compensable.compensationMethod().isEmpty() ? "" : compensationMethodSignature(joinPoint, compensable, method); String retrySignature = (retries != 0 || compensationSignature.isEmpty()) ? method.toString() : ""; boolean isProceed = false; boolean enabledTx = false; try { CurrentThreadOmegaContext.putThreadGlobalLocalTxId(new OmegaContextServiceConfig(context, false, false)); AlphaResponse response = interceptor.preIntercept(parentTxId, compensationSignature, compensable.timeout(), retrySignature, retries, joinPoint.getArgs()); enabledTx = response.enabledTx(); isProceed = true; Object result = null; if (!response.aborted()) { result = joinPoint.proceed(); } if (enabledTx) { if (response.aborted()) { String abortedLocalTxId = context.localTxId(); context.setLocalTxId(parentTxId); throw new InvalidTransactionException("Abort sub transaction " + abortedLocalTxId + " because global transaction " + context.globalTxId() + " has already aborted."); } CurrentThreadOmegaContext.clearCache(); interceptor.postIntercept(parentTxId, compensationSignature); } return result; } catch (InvalidTransactionException ite) { throw ite; } catch (Throwable throwable) { boolean isFaultTolerant = ApplicationContextUtil.getApplicationContext().getBean(MessageSender.class).readConfigFromServer(ConfigCenterType.CompensationFaultTolerant.toInteger(), context.category()).getStatus(); if (enabledTx && !isFaultTolerant) { interceptor.onError(parentTxId, compensationSignature, throwable); } if (!isProceed && isFaultTolerant) { return joinPoint.proceed(); } throw throwable; } } }
DefaultRecovery implements RecoveryPolicy { @Override public Object apply(ProceedingJoinPoint joinPoint, Compensable compensable, CompensableInterceptor interceptor, OmegaContext context, String parentTxId, int retries) throws Throwable { Method method = ((MethodSignature) joinPoint.getSignature()).getMethod(); LOG.debug("Intercepting compensable method {} with context {}", method.toString(), context); String compensationSignature = compensable.compensationMethod().isEmpty() ? "" : compensationMethodSignature(joinPoint, compensable, method); String retrySignature = (retries != 0 || compensationSignature.isEmpty()) ? method.toString() : ""; boolean isProceed = false; boolean enabledTx = false; try { CurrentThreadOmegaContext.putThreadGlobalLocalTxId(new OmegaContextServiceConfig(context, false, false)); AlphaResponse response = interceptor.preIntercept(parentTxId, compensationSignature, compensable.timeout(), retrySignature, retries, joinPoint.getArgs()); enabledTx = response.enabledTx(); isProceed = true; Object result = null; if (!response.aborted()) { result = joinPoint.proceed(); } if (enabledTx) { if (response.aborted()) { String abortedLocalTxId = context.localTxId(); context.setLocalTxId(parentTxId); throw new InvalidTransactionException("Abort sub transaction " + abortedLocalTxId + " because global transaction " + context.globalTxId() + " has already aborted."); } CurrentThreadOmegaContext.clearCache(); interceptor.postIntercept(parentTxId, compensationSignature); } return result; } catch (InvalidTransactionException ite) { throw ite; } catch (Throwable throwable) { boolean isFaultTolerant = ApplicationContextUtil.getApplicationContext().getBean(MessageSender.class).readConfigFromServer(ConfigCenterType.CompensationFaultTolerant.toInteger(), context.category()).getStatus(); if (enabledTx && !isFaultTolerant) { interceptor.onError(parentTxId, compensationSignature, throwable); } if (!isProceed && isFaultTolerant) { return joinPoint.proceed(); } throw throwable; } }  }
DefaultRecovery implements RecoveryPolicy { @Override public Object apply(ProceedingJoinPoint joinPoint, Compensable compensable, CompensableInterceptor interceptor, OmegaContext context, String parentTxId, int retries) throws Throwable { Method method = ((MethodSignature) joinPoint.getSignature()).getMethod(); LOG.debug("Intercepting compensable method {} with context {}", method.toString(), context); String compensationSignature = compensable.compensationMethod().isEmpty() ? "" : compensationMethodSignature(joinPoint, compensable, method); String retrySignature = (retries != 0 || compensationSignature.isEmpty()) ? method.toString() : ""; boolean isProceed = false; boolean enabledTx = false; try { CurrentThreadOmegaContext.putThreadGlobalLocalTxId(new OmegaContextServiceConfig(context, false, false)); AlphaResponse response = interceptor.preIntercept(parentTxId, compensationSignature, compensable.timeout(), retrySignature, retries, joinPoint.getArgs()); enabledTx = response.enabledTx(); isProceed = true; Object result = null; if (!response.aborted()) { result = joinPoint.proceed(); } if (enabledTx) { if (response.aborted()) { String abortedLocalTxId = context.localTxId(); context.setLocalTxId(parentTxId); throw new InvalidTransactionException("Abort sub transaction " + abortedLocalTxId + " because global transaction " + context.globalTxId() + " has already aborted."); } CurrentThreadOmegaContext.clearCache(); interceptor.postIntercept(parentTxId, compensationSignature); } return result; } catch (InvalidTransactionException ite) { throw ite; } catch (Throwable throwable) { boolean isFaultTolerant = ApplicationContextUtil.getApplicationContext().getBean(MessageSender.class).readConfigFromServer(ConfigCenterType.CompensationFaultTolerant.toInteger(), context.category()).getStatus(); if (enabledTx && !isFaultTolerant) { interceptor.onError(parentTxId, compensationSignature, throwable); } if (!isProceed && isFaultTolerant) { return joinPoint.proceed(); } throw throwable; } }  @Override Object apply(ProceedingJoinPoint joinPoint, Compensable compensable, CompensableInterceptor interceptor,
      OmegaContext context, String parentTxId, int retries); }
DefaultRecovery implements RecoveryPolicy { @Override public Object apply(ProceedingJoinPoint joinPoint, Compensable compensable, CompensableInterceptor interceptor, OmegaContext context, String parentTxId, int retries) throws Throwable { Method method = ((MethodSignature) joinPoint.getSignature()).getMethod(); LOG.debug("Intercepting compensable method {} with context {}", method.toString(), context); String compensationSignature = compensable.compensationMethod().isEmpty() ? "" : compensationMethodSignature(joinPoint, compensable, method); String retrySignature = (retries != 0 || compensationSignature.isEmpty()) ? method.toString() : ""; boolean isProceed = false; boolean enabledTx = false; try { CurrentThreadOmegaContext.putThreadGlobalLocalTxId(new OmegaContextServiceConfig(context, false, false)); AlphaResponse response = interceptor.preIntercept(parentTxId, compensationSignature, compensable.timeout(), retrySignature, retries, joinPoint.getArgs()); enabledTx = response.enabledTx(); isProceed = true; Object result = null; if (!response.aborted()) { result = joinPoint.proceed(); } if (enabledTx) { if (response.aborted()) { String abortedLocalTxId = context.localTxId(); context.setLocalTxId(parentTxId); throw new InvalidTransactionException("Abort sub transaction " + abortedLocalTxId + " because global transaction " + context.globalTxId() + " has already aborted."); } CurrentThreadOmegaContext.clearCache(); interceptor.postIntercept(parentTxId, compensationSignature); } return result; } catch (InvalidTransactionException ite) { throw ite; } catch (Throwable throwable) { boolean isFaultTolerant = ApplicationContextUtil.getApplicationContext().getBean(MessageSender.class).readConfigFromServer(ConfigCenterType.CompensationFaultTolerant.toInteger(), context.category()).getStatus(); if (enabledTx && !isFaultTolerant) { interceptor.onError(parentTxId, compensationSignature, throwable); } if (!isProceed && isFaultTolerant) { return joinPoint.proceed(); } throw throwable; } }  @Override Object apply(ProceedingJoinPoint joinPoint, Compensable compensable, CompensableInterceptor interceptor,
      OmegaContext context, String parentTxId, int retries);  }