You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@servicecomb.apache.org by ni...@apache.org on 2018/08/20 11:11:44 UTC
[incubator-servicecomb-saga] 02/05: SCB-817 Added TCC events in
Omega part (WIP)
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 0c1663454afb885308c0b035a918589204c64256
Author: Willem Jiang <ji...@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 {
-}