This is an automated email from the ASF dual-hosted git repository. ningjiang pushed a commit to branch SCB-665 in repository https://gitbox.apache.org/repos/asf/incubator-servicecomb-saga.git
commit 35d946c7f94680aa6d8a257dc2b9165353737413 Author: Willem Jiang <jiangni...@huawei.com> AuthorDate: Thu Aug 16 10:44:06 2018 +0800 SCB-817 Added TCC events in Omega part (WIP) --- .../saga/omega/context/annotations/TccStart.java | 7 +++ .../annotations/{TCC.java => Participate.java} | 14 ++--- .../saga/omega/transaction/tcc/TccAspect.java | 48 --------------- .../saga/omega/transaction/tcc/TccInterceptor.java | 4 -- .../transaction/tcc/TccParticipatorAspect.java | 68 ++++++++++++++++++++++ .../tcc/TccStartAnnotationProcessor.java | 6 +- .../saga/omega/transaction/tcc/TccStartAspect.java | 7 ++- .../omega/transaction/tcc/events/CancelEvent.java | 20 ------- .../omega/transaction/tcc/events/ConfirmEvent.java | 20 ------- .../omega/transaction/tcc/events/TccEndEvent.java | 14 +++++ .../transaction/tcc/events/TccStartEvent.java | 25 ++++++++ .../tcc/events/TransactionEndEvent.java | 4 -- .../tcc/events/TransationStartEvent.java | 4 -- 13 files changed, 129 insertions(+), 112 deletions(-) diff --git a/omega/omega-context/src/main/java/org/apache/servicecomb/saga/omega/context/annotations/TccStart.java b/omega/omega-context/src/main/java/org/apache/servicecomb/saga/omega/context/annotations/TccStart.java index c9154f9..6c8afd2 100644 --- a/omega/omega-context/src/main/java/org/apache/servicecomb/saga/omega/context/annotations/TccStart.java +++ b/omega/omega-context/src/main/java/org/apache/servicecomb/saga/omega/context/annotations/TccStart.java @@ -12,4 +12,11 @@ import java.lang.annotation.Target; @Retention(RUNTIME) @Target(METHOD) public @interface TccStart { + /** + * TCC timeout, in seconds. <br> + * Default value is 0, which means never timeout. + * + * @return + */ + int timeout() default 0; } diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/annotations/TCC.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/annotations/Participate.java similarity index 79% rename from omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/annotations/TCC.java rename to omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/annotations/Participate.java index a095978..8351076 100644 --- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/annotations/TCC.java +++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/annotations/Participate.java @@ -25,22 +25,22 @@ import java.lang.annotation.Target; @Target({ElementType.METHOD}) /** * Indicates the annotated method will start a sub-transaction. <br> - * A <code>@TCC</code> method should satisfy below requirements: + * A <code>@Participate</code> method should satisfy below requirements: * <ol> * <li>all parameters are serialized</li> * <li>is idempotent</li> - * <li>the object instance which @TCC method resides in should be stateless</li> + * <li>the object instance which @Participate method resides in should be stateless</li> * </ol> */ -public @interface TCC { +public @interface Participate { /** * Confirm method name.<br> * A confirm method should satisfy below requirements: * <ol> - * <li>has same parameter list as @TCC method's</li> + * <li>has same parameter list as @Participate method's</li> * <li>all parameters are serialized</li> * <li>is idempotent</li> - * <li>be in the same class as @TCC method is in</li> + * <li>be in the same class as @Participate method is in</li> * </ol> * * @return @@ -51,10 +51,10 @@ public @interface TCC { * Cancel method name.<br> * A cancel method should satisfy below requirements: * <ol> - * <li>has same parameter list as @TCC method's</li> + * <li>has same parameter list as @Participate method's</li> * <li>all parameters are serialized</li> * <li>is idempotent</li> - * <li>be in the same class as @TCC method is in</li> + * <li>be in the same class as @Participate method is in</li> * </ol> * * @return diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccAspect.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccAspect.java deleted file mode 100644 index 173cb78..0000000 --- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccAspect.java +++ /dev/null @@ -1,48 +0,0 @@ -package org.apache.servicecomb.saga.omega.transaction.tcc; - -import java.lang.invoke.MethodHandles; -import java.lang.reflect.Method; - -import org.apache.servicecomb.saga.omega.context.OmegaContext; -import org.apache.servicecomb.saga.omega.transaction.CompensableInterceptor; -import org.apache.servicecomb.saga.omega.transaction.MessageSender; -import org.apache.servicecomb.saga.omega.transaction.RecoveryPolicy; -import org.apache.servicecomb.saga.omega.transaction.RecoveryPolicyFactory; -import org.apache.servicecomb.saga.omega.transaction.annotations.Compensable; -import org.aspectj.lang.ProceedingJoinPoint; -import org.aspectj.lang.annotation.Around; -import org.aspectj.lang.annotation.Aspect; -import org.aspectj.lang.reflect.MethodSignature; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -@Aspect -public class TccAspect { - private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); - - private final OmegaContext context; - - private final CompensableInterceptor interceptor; - - public TccAspect(MessageSender sender, OmegaContext context) { - this.context = context; - this.interceptor = new CompensableInterceptor(context, sender); - } - - @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); - } finally { - context.setLocalTxId(localTxId); - LOG.debug("Restored context back to {}", context); - } - } -} diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccInterceptor.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccInterceptor.java deleted file mode 100644 index 5ada269..0000000 --- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccInterceptor.java +++ /dev/null @@ -1,4 +0,0 @@ -package org.apache.servicecomb.saga.omega.transaction.tcc; - -public class TccInterceptor { -} diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccParticipatorAspect.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccParticipatorAspect.java new file mode 100644 index 0000000..02adac2 --- /dev/null +++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccParticipatorAspect.java @@ -0,0 +1,68 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.servicecomb.saga.omega.transaction.tcc; + +import java.lang.invoke.MethodHandles; +import java.lang.reflect.Method; + +import org.apache.servicecomb.saga.omega.context.OmegaContext; +import org.apache.servicecomb.saga.omega.transaction.MessageSender; +import org.apache.servicecomb.saga.omega.transaction.OmegaException; +import org.apache.servicecomb.saga.omega.transaction.RecoveryPolicy; +import org.apache.servicecomb.saga.omega.transaction.RecoveryPolicyFactory; +import org.apache.servicecomb.saga.omega.transaction.annotations.Compensable; +import org.apache.servicecomb.saga.omega.transaction.annotations.Participate; +import org.aspectj.lang.ProceedingJoinPoint; +import org.aspectj.lang.annotation.Around; +import org.aspectj.lang.annotation.Aspect; +import org.aspectj.lang.reflect.MethodSignature; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +@Aspect +public class TccParticipatorAspect { + private static final Logger LOG = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass()); + + private final OmegaContext context; + + public TccParticipatorAspect(MessageSender sender, OmegaContext context) { + this.context = context; + } + + @Around("execution(@org.apache.servicecomb.saga.omega.transaction.annotations.Participate * *(..)) && @annotation(participate)") + Object advise(ProceedingJoinPoint joinPoint, Participate participate) throws Throwable { + Method method = ((MethodSignature) joinPoint.getSignature()).getMethod(); + String localTxId = context.localTxId(); + String cancelMethod = participate.cancelMethod(); + + context.newLocalTxId(); + LOG.debug("Updated context {} for participate method {} ", context, method.toString()); + + try { + Object result = joinPoint.proceed(); + // Send the participate message back + LOG.debug("Participate Transaction with context {} has finished.", context); + + return result; + } catch (Throwable throwable) { + // Now we don't handle the error message + LOG.error("Participate Transaction with context {} failed.", context, throwable); + throw throwable; + } + } +} diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAnnotationProcessor.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAnnotationProcessor.java index 3e708cd..b38d876 100644 --- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAnnotationProcessor.java +++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAnnotationProcessor.java @@ -33,14 +33,16 @@ public class TccStartAnnotationProcessor implements EventAwareInterceptor { @Override public void postIntercept(String parentTxId, String compensationMethod) { - AlphaResponse response = sender.send(new SagaEndedEvent(omegaContext.globalTxId(), omegaContext.localTxId())); + // Send the confirm event + /*AlphaResponse response = sender.send(new SagaEndedEvent(omegaContext.globalTxId(), omegaContext.localTxId())); if (response.aborted()) { throw new OmegaException("transaction " + parentTxId + " is aborted"); - } + }*/ } @Override public void onError(String parentTxId, String compensationMethod, Throwable throwable) { + // Send the cancel event String globalTxId = omegaContext.globalTxId(); sender.send(new TxAbortedEvent(globalTxId, omegaContext.localTxId(), null, compensationMethod, throwable)); } diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAspect.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAspect.java index 5b380db..c8b7936 100644 --- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAspect.java +++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/TccStartAspect.java @@ -5,6 +5,7 @@ import java.lang.reflect.Method; import org.apache.servicecomb.saga.omega.context.OmegaContext; import org.apache.servicecomb.saga.omega.context.annotations.SagaStart; +import org.apache.servicecomb.saga.omega.context.annotations.TccStart; import org.apache.servicecomb.saga.omega.transaction.MessageSender; import org.apache.servicecomb.saga.omega.transaction.OmegaException; import org.aspectj.lang.ProceedingJoinPoint; @@ -27,12 +28,12 @@ public class TccStartAspect { this.tccStartAnnotationProcessor = new TccStartAnnotationProcessor(context, sender); } - @Around("execution(@org.apache.servicecomb.saga.omega.context.annotations.TccStart * *(..)) && @annotation(sagaStart)") - Object advise(ProceedingJoinPoint joinPoint, SagaStart sagaStart) throws Throwable { + @Around("execution(@org.apache.servicecomb.saga.omega.context.annotations.TccStart * *(..)) && @annotation(tccStart)") + Object advise(ProceedingJoinPoint joinPoint, TccStart tccStart) throws Throwable { initializeOmegaContext(); Method method = ((MethodSignature) joinPoint.getSignature()).getMethod(); - tccStartAnnotationProcessor.preIntercept(context.globalTxId(), method.toString(), sagaStart.timeout(), "", 0); + tccStartAnnotationProcessor.preIntercept(context.globalTxId(), method.toString(), tccStart.timeout(), "", 0); LOG.debug("Initialized context {} before execution of method {}", context, method.toString()); try { diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/CancelEvent.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/CancelEvent.java deleted file mode 100644 index 7659812..0000000 --- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/CancelEvent.java +++ /dev/null @@ -1,20 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.servicecomb.saga.omega.transaction.tcc.events; - -public class CancelEvent { -} diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/ConfirmEvent.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/ConfirmEvent.java deleted file mode 100644 index f01675b..0000000 --- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/ConfirmEvent.java +++ /dev/null @@ -1,20 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.servicecomb.saga.omega.transaction.tcc.events; - -public class ConfirmEvent { -} diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/TccEndEvent.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/TccEndEvent.java new file mode 100644 index 0000000..1022aee --- /dev/null +++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/TccEndEvent.java @@ -0,0 +1,14 @@ +package org.apache.servicecomb.saga.omega.transaction.tcc.events; + +public class TccEndEvent { + private final String globalTxId; + private final String localTxId; + private final String parentTxId; + + + public TccEndEvent(String globalTxId, String localTxId, String parentTxId) { + this.globalTxId = globalTxId; + this.localTxId = localTxId; + this.parentTxId = parentTxId; + } +} diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/TccStartEvent.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/TccStartEvent.java new file mode 100644 index 0000000..c94dbca --- /dev/null +++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/TccStartEvent.java @@ -0,0 +1,25 @@ +package org.apache.servicecomb.saga.omega.transaction.tcc.events; + +public class TccStartEvent { + private final String globalTxId; + private final String localTxId; + private final String parentTxId; + + public String getGlobalTxId() { + return globalTxId; + } + + public String getLocalTxId() { + return localTxId; + } + + public String getParentTxId() { + return parentTxId; + } + + public TccStartEvent(String globalTxId, String localTxId, String parentTxId) { + this.globalTxId = globalTxId; + this.localTxId = localTxId; + this.parentTxId = parentTxId; + } +} diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/TransactionEndEvent.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/TransactionEndEvent.java deleted file mode 100644 index 9613602..0000000 --- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/TransactionEndEvent.java +++ /dev/null @@ -1,4 +0,0 @@ -package org.apache.servicecomb.saga.omega.transaction.tcc.events; - -public class TransactionEndEvent { -} diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/TransationStartEvent.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/TransationStartEvent.java deleted file mode 100644 index 54858c6..0000000 --- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/saga/omega/transaction/tcc/events/TransationStartEvent.java +++ /dev/null @@ -1,4 +0,0 @@ -package org.apache.servicecomb.saga.omega.transaction.tcc.events; - -public class TransationStartEvent { -}