[
https://issues.apache.org/jira/browse/SCB-877?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16614849#comment-16614849
]
ASF GitHub Bot commented on SCB-877:
------------------------------------
WillemJiang closed pull request #295: SCB-877 persisted the received TCC events
URL: https://github.com/apache/incubator-servicecomb-saga/pull/295
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/AlphaConfig.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/AlphaConfig.java
index 585f7d9f..4d724676 100644
---
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/AlphaConfig.java
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/AlphaConfig.java
@@ -37,7 +37,7 @@
import org.apache.servicecomb.saga.alpha.server.tcc.GrpcTccEventService;
import
org.apache.servicecomb.saga.alpha.server.tcc.callback.OmegaCallbackWrapper;
import org.apache.servicecomb.saga.alpha.server.tcc.callback.TccCallbackEngine;
-import org.apache.servicecomb.saga.alpha.server.tcc.TransactionEventService;
+import org.apache.servicecomb.saga.alpha.server.tcc.TccTxEventFacade;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.autoconfigure.domain.EntityScan;
@@ -102,23 +102,16 @@ TxConsistentService txConsistentService(
}
@Bean
- TransactionEventService transactionEventService(
+ TccTxEventFacade tccTxEventFacade(
@Value("${alpha.server.storage:rdb}") String storage,
- @Qualifier("defaultTransactionEventService") TransactionEventService
defaultTransactionEventService,
- @Qualifier("rdbTransactionEventService") TransactionEventService
rdbTransactionEventService) {
- return "rdb".equals(storage) ? rdbTransactionEventService :
defaultTransactionEventService;
+ @Qualifier("defaultTccTxEventFacade") TccTxEventFacade
defaultTccTxEventFacade,
+ @Qualifier("rdbTccTxEventFacade") TccTxEventFacade rdbTccTxEventFacade) {
+ return "rdb".equals(storage) ? rdbTccTxEventFacade :
defaultTccTxEventFacade;
}
@Bean
- TccCallbackEngine tccCallbackEngine(TransactionEventService
transactionEventService) {
- return new TccCallbackEngine(new OmegaCallbackWrapper(),
transactionEventService);
- }
-
- @Bean
- GrpcTccEventService grpcTccEventService(
- TransactionEventService transactionEventService,
- TccCallbackEngine tccCallbackEngine) {
- return new GrpcTccEventService(tccCallbackEngine, transactionEventService);
+ GrpcTccEventService grpcTccEventService(TccTxEventFacade tccTxEventFacade) {
+ return new GrpcTccEventService(tccTxEventFacade);
}
@Bean
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/DefaultTransactionEventService.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/DefaultTccTxEventFacadeImpl.java
similarity index 50%
rename from
alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/DefaultTransactionEventService.java
rename to
alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/DefaultTccTxEventFacadeImpl.java
index 9f119f40..e554fc7d 100644
---
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/DefaultTransactionEventService.java
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/DefaultTccTxEventFacadeImpl.java
@@ -18,26 +18,38 @@
package org.apache.servicecomb.saga.alpha.server.tcc;
import java.lang.invoke.MethodHandles;
-import java.util.LinkedHashSet;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-
+import java.util.List;
+import org.apache.servicecomb.saga.alpha.server.tcc.callback.TccCallbackEngine;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEvent;
import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
-import org.apache.servicecomb.saga.common.TransactionStatus;
+import
org.apache.servicecomb.saga.alpha.server.tcc.service.MemoryEventRegistry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;
/**
* Manage TCC transaction event.
*/
-@Component("defaultTransactionEventService")
-public final class DefaultTransactionEventService implements
TransactionEventService {
+@Component("defaultTccTxEventFacade")
+public final class DefaultTccTxEventFacadeImpl implements TccTxEventFacade {
private static final Logger LOG =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private final Map<String, Set<ParticipatedEvent>> REGISTRY = new
ConcurrentHashMap<>();
+ @Autowired
+ @Qualifier("memoryCallbackEngine")
+ private TccCallbackEngine tccCallbackEngine;
+
+ @Override
+ public boolean onTccStartEvent(GlobalTxEvent globalTxEvent) {
+ MemoryEventRegistry.addGlobalTxEvent(globalTxEvent);
+ LOG.info("Registered participated event, global tx: {}, local tx: {},
parent id: {}, "
+ + "txType: {}, service [{}] instanceId [{}]",
+ globalTxEvent.getGlobalTxId(), globalTxEvent.getLocalTxId(),
globalTxEvent.getParentTxId(),
+ globalTxEvent.getTxType(), globalTxEvent.getServiceName(),
globalTxEvent.getInstanceId());
+ return true;
+ }
/**
* Register participate event.
@@ -46,12 +58,8 @@
*/
@Override
- public boolean addEvent(ParticipatedEvent participatedEvent) {
-
- REGISTRY
- .computeIfAbsent(participatedEvent.getGlobalTxId(), key -> new
LinkedHashSet<>())
- .add(participatedEvent);
-
+ public boolean onParticipateEvent(ParticipatedEvent participatedEvent) {
+ MemoryEventRegistry.addParticipateEvent(participatedEvent);
LOG.info("Registered participated event, global tx: {}, local tx: {},
parent id: {}, "
+ "confirm: {}, cancel: {}, status: {}, service [{}] instanceId
[{}]",
participatedEvent.getGlobalTxId(), participatedEvent.getLocalTxId(),
participatedEvent.getParentTxId(),
@@ -62,6 +70,22 @@ public boolean addEvent(ParticipatedEvent participatedEvent)
{
return true;
}
+ @Override
+ public boolean onTccEndEvent(GlobalTxEvent globalTxEvent) {
+ MemoryEventRegistry.addGlobalTxEvent(globalTxEvent);
+ return tccCallbackEngine.execute(globalTxEvent);
+ }
+
+ @Override
+ public void onCoordinatedEvent(String globalTxId, String localTxId) {
+ MemoryEventRegistry.migrateParticipate(globalTxId, localTxId);
+ }
+
+ @Override
+ public List<GlobalTxEvent> getGlobalTxEventByGlobalTxId(String globalTxId) {
+ return MemoryEventRegistry.getGlobalTxEventByGlobalTxId(globalTxId);
+ }
+
/**
* Retrieve participate event from registry.
*
@@ -69,11 +93,12 @@ public boolean addEvent(ParticipatedEvent
participatedEvent) {
* @return participate events
*/
@Override
- public Set<ParticipatedEvent> getEventByGlobalTxId(String globalTxId) {
- return REGISTRY.get(globalTxId);
+ public List<ParticipatedEvent> getParticipateEventByGlobalTxId(String
globalTxId) {
+ return MemoryEventRegistry.getParticipateEventByGlobalTxId(globalTxId);
}
@Override
- public void migration(String globalTxId, String localTxId) {
+ public void migrationGlobalTxEvent(String globalTxId, String localTxId) {
+ MemoryEventRegistry.migrationGlobalTxEvent(globalTxId, localTxId);
}
}
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/GrpcTccEventService.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/GrpcTccEventService.java
index d3dad677..9fe2f2cf 100644
---
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/GrpcTccEventService.java
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/GrpcTccEventService.java
@@ -20,9 +20,8 @@
import io.grpc.stub.StreamObserver;
import java.lang.invoke.MethodHandles;
import org.apache.servicecomb.saga.alpha.server.tcc.callback.OmegaCallback;
-import org.apache.servicecomb.saga.alpha.server.tcc.callback.TccCallbackEngine;
-import
org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEventFactory;
import
org.apache.servicecomb.saga.alpha.server.tcc.callback.OmegaCallbacksRegistry;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TxEventFactory;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcAck;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcServiceConfig;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTccCoordinateCommand;
@@ -45,14 +44,10 @@
private static final GrpcAck REJECT =
GrpcAck.newBuilder().setAborted(true).build();
- private final TccCallbackEngine tccCallbackEngine;
+ private final TccTxEventFacade tccTxEventFacade;
- private final TransactionEventService transactionEventService;
-
- public GrpcTccEventService(TccCallbackEngine tccCallbackEngine,
- TransactionEventService transactionEventService) {
- this.tccCallbackEngine = tccCallbackEngine;
- this.transactionEventService = transactionEventService;
+ public GrpcTccEventService(TccTxEventFacade tccTxEventFacade) {
+ this.tccTxEventFacade = tccTxEventFacade;
}
@Override
@@ -64,7 +59,7 @@ public void onConnected(GrpcServiceConfig request,
StreamObserver<GrpcTccCoordin
@Override
public void onTccTransactionStarted(GrpcTccTransactionStartedEvent request,
StreamObserver<GrpcAck> responseObserver) {
LOG.info("Received transaction start event, global tx id: {}",
request.getGlobalTxId());
- responseObserver.onNext(ALLOW);
+
responseObserver.onNext(tccTxEventFacade.onTccStartEvent(TxEventFactory.create(request))
? ALLOW : REJECT);
responseObserver.onCompleted();
}
@@ -72,15 +67,14 @@ public void
onTccTransactionStarted(GrpcTccTransactionStartedEvent request, Stre
public void participate(GrpcTccParticipatedEvent request,
StreamObserver<GrpcAck> responseObserver) {
LOG.info("Received participated event from service {} , global tx id: {},
local tx id: {}", request.getServiceName(),
request.getGlobalTxId(), request.getLocalTxId()) ;
- boolean ok =
transactionEventService.addEvent(ParticipatedEventFactory.create(request));
- responseObserver.onNext(ok ? ALLOW : REJECT);
+
responseObserver.onNext(tccTxEventFacade.onParticipateEvent(TxEventFactory.create(request))
? ALLOW : REJECT);
responseObserver.onCompleted();
}
@Override
public void onTccTransactionEnded(GrpcTccTransactionEndedEvent request,
StreamObserver<GrpcAck> responseObserver) {
LOG.info("Received transaction end event, global tx id: {}",
request.getGlobalTxId());
- responseObserver.onNext(tccCallbackEngine.execute(request) ? ALLOW :
REJECT);
+
responseObserver.onNext(tccTxEventFacade.onTccEndEvent(TxEventFactory.create(request))
? ALLOW : REJECT);
responseObserver.onCompleted();
}
@@ -90,7 +84,7 @@ public void onTccCoordinated(GrpcTccCoordinatedEvent request,
StreamObserver<Grp
+ "method: {}, status: {}, service [{}] instanceId [{}]",
request.getGlobalTxId(), request.getLocalTxId(),
request.getParentTxId(),
request.getMethodName(), request.getStatus(),
request.getServiceName(), request.getInstanceId());
- transactionEventService.migration(request.getGlobalTxId(),
request.getLocalTxId());
+ tccTxEventFacade.onCoordinatedEvent(request.getGlobalTxId(),
request.getLocalTxId());
responseObserver.onNext(ALLOW);
responseObserver.onCompleted();
}
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/RdbTccTxEventFacadeImpl.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/RdbTccTxEventFacadeImpl.java
new file mode 100644
index 00000000..7978fe07
--- /dev/null
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/RdbTccTxEventFacadeImpl.java
@@ -0,0 +1,86 @@
+/*
+ * 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.alpha.server.tcc;
+
+import com.google.common.collect.Lists;
+import java.lang.invoke.MethodHandles;
+import java.util.List;
+import org.apache.servicecomb.saga.alpha.server.tcc.callback.TccCallbackEngine;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
+import
org.apache.servicecomb.saga.alpha.server.tcc.service.GlobalTxEventService;
+import
org.apache.servicecomb.saga.alpha.server.tcc.service.ParticipateEventService;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.stereotype.Component;
+
+@Component("rdbTccTxEventFacade")
+public class RdbTccTxEventFacadeImpl implements TccTxEventFacade {
+
+ @Autowired
+ private ParticipateEventService participateEventService;
+
+ @Autowired
+ private GlobalTxEventService globalTxEventService;
+
+ @Autowired
+ @Qualifier("rdbCallbackEngine")
+ private TccCallbackEngine tccCallbackEngine;
+
+ private static final Logger LOG =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ @Override
+ public boolean onTccStartEvent(GlobalTxEvent globalTxEvent) {
+ return globalTxEventService.addEvent(globalTxEvent);
+ }
+
+ @Override
+ public boolean onParticipateEvent(ParticipatedEvent event) {
+ return participateEventService.addEvent(event);
+ }
+
+ @Override
+ public boolean onTccEndEvent(GlobalTxEvent globalTxEvent) {
+ if (globalTxEventService.addEvent(globalTxEvent)) {
+ return tccCallbackEngine.execute(globalTxEvent);
+ }
+ return false;
+ }
+
+ @Override
+ public void onCoordinatedEvent(String globalTxId, String localTxId) {
+ participateEventService.migration(globalTxId, localTxId);
+ }
+
+ @Override
+ public List<GlobalTxEvent> getGlobalTxEventByGlobalTxId(String globalTxId) {
+ return
globalTxEventService.getEventByGlobalTxId(globalTxId).orElse(Lists.newArrayList());
+ }
+
+ @Override
+ public List<ParticipatedEvent> getParticipateEventByGlobalTxId(String
globalTxId) {
+ return
participateEventService.getEventByGlobalTxId(globalTxId).orElse(Lists.newArrayList());
+ }
+
+ @Override
+ public void migrationGlobalTxEvent(String globalTxId, String localTxId) {
+ globalTxEventService.migration(globalTxId, localTxId);
+ }
+}
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/TccTxEventFacade.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/TccTxEventFacade.java
new file mode 100644
index 00000000..64209fec
--- /dev/null
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/TccTxEventFacade.java
@@ -0,0 +1,41 @@
+/*
+ * 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.alpha.server.tcc;
+
+import java.util.List;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
+
+public interface TccTxEventFacade {
+
+ boolean onTccStartEvent(GlobalTxEvent globalTxEvent);
+
+ boolean onParticipateEvent(ParticipatedEvent participateEvent);
+
+ boolean onTccEndEvent(GlobalTxEvent globalTxEvent);
+
+ void onCoordinatedEvent(String globalTxId, String localTxId);
+
+ List<GlobalTxEvent> getGlobalTxEventByGlobalTxId(String globalTxId);
+
+ List<ParticipatedEvent> getParticipateEventByGlobalTxId(String globalTxId);
+
+// void migrationParticipateEvent(String globalTxId, String localTxId);
+//
+ void migrationGlobalTxEvent(String globalTxId, String localTxId);
+}
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/CallbackEngine.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/CallbackEngine.java
index a8521e69..f262f52c 100644
---
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/CallbackEngine.java
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/CallbackEngine.java
@@ -17,9 +17,9 @@
package org.apache.servicecomb.saga.alpha.server.tcc.callback;
-import
org.apache.servicecomb.saga.pack.contract.grpc.GrpcTccTransactionEndedEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEvent;
public interface CallbackEngine {
- boolean execute(GrpcTccTransactionEndedEvent request);
+ boolean execute(GlobalTxEvent request);
}
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/MemoryCallbackEngine.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/MemoryCallbackEngine.java
new file mode 100644
index 00000000..5a570c68
--- /dev/null
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/MemoryCallbackEngine.java
@@ -0,0 +1,32 @@
+/*
+ * 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.alpha.server.tcc.callback;
+
+import java.util.List;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
+import
org.apache.servicecomb.saga.alpha.server.tcc.service.MemoryEventRegistry;
+import org.springframework.stereotype.Component;
+
+@Component("memoryCallbackEngine")
+public class MemoryCallbackEngine extends TccCallbackEngine {
+
+ @Override
+ protected List<ParticipatedEvent> findParticipate(String globalTxId) {
+ return MemoryEventRegistry.getParticipateEventByGlobalTxId(globalTxId);
+ }
+}
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/OmegaCallbackWrapper.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/OmegaCallbackWrapper.java
index fdda5338..64d624a2 100644
---
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/OmegaCallbackWrapper.java
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/OmegaCallbackWrapper.java
@@ -19,7 +19,9 @@
import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
import org.apache.servicecomb.saga.common.TransactionStatus;
+import org.springframework.stereotype.Component;
+@Component
public class OmegaCallbackWrapper implements OmegaCallback {
@Override
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/RdbCallbackEngine.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/RdbCallbackEngine.java
new file mode 100644
index 00000000..bf3ee035
--- /dev/null
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/RdbCallbackEngine.java
@@ -0,0 +1,37 @@
+/*
+ * 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.alpha.server.tcc.callback;
+
+import com.google.common.collect.Lists;
+import java.util.List;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
+import
org.apache.servicecomb.saga.alpha.server.tcc.service.ParticipateEventService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+@Component("rdbCallbackEngine")
+public class RdbCallbackEngine extends TccCallbackEngine {
+
+ @Autowired
+ private ParticipateEventService participateEventService;
+
+ @Override
+ protected List<ParticipatedEvent> findParticipate(String globalTxId) {
+ return
participateEventService.getEventByGlobalTxId(globalTxId).orElse(Lists.newArrayList());
+ }
+}
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/TccCallbackEngine.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/TccCallbackEngine.java
index 9b6ca599..9cb29dd7 100644
---
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/TccCallbackEngine.java
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/callback/TccCallbackEngine.java
@@ -18,32 +18,26 @@
package org.apache.servicecomb.saga.alpha.server.tcc.callback;
import java.lang.invoke.MethodHandles;
+import java.util.List;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEvent;
import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
-import org.apache.servicecomb.saga.alpha.server.tcc.TransactionEventService;
import org.apache.servicecomb.saga.common.TransactionStatus;
-import
org.apache.servicecomb.saga.pack.contract.grpc.GrpcTccTransactionEndedEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
-public class TccCallbackEngine implements CallbackEngine {
+public abstract class TccCallbackEngine implements CallbackEngine {
private static final Logger LOG =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
- private final OmegaCallbackWrapper omegaCallbackWrapper;
-
- private final TransactionEventService transactionEventService;
-
- public TccCallbackEngine(
- OmegaCallbackWrapper omegaCallbackWrapper,
- TransactionEventService transactionEventService) {
- this.omegaCallbackWrapper = omegaCallbackWrapper;
- this.transactionEventService = transactionEventService;
- }
+ @Autowired
+ private OmegaCallbackWrapper omegaCallbackWrapper;
@Override
- public boolean execute(GrpcTccTransactionEndedEvent request) {
+ public boolean execute(GlobalTxEvent request) {
boolean result = true;
- for (ParticipatedEvent event :
transactionEventService.getEventByGlobalTxId(request.getGlobalTxId())) {
+ List<ParticipatedEvent> list = findParticipate(request.getGlobalTxId());
+ for (ParticipatedEvent event : list) {
try {
// only invoke the event is succeed
if (event.getStatus().equals(TransactionStatus.Succeed.toString())) {
@@ -57,6 +51,8 @@ public boolean execute(GrpcTccTransactionEndedEvent request) {
return result;
}
+ protected abstract List<ParticipatedEvent> findParticipate(String
globalTxId);
+
private void logError(ParticipatedEvent event, Exception ex) {
LOG.error(
"Failed to invoke service [{}] instance [{}] with method [{}], global
tx id [{}] and local tx id [{}]",
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/GlobalTxEvent.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/GlobalTxEvent.java
new file mode 100644
index 00000000..bb3ff8b1
--- /dev/null
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/GlobalTxEvent.java
@@ -0,0 +1,124 @@
+/*
+ * 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.alpha.server.tcc.jpa;
+
+import java.util.Date;
+import java.util.Objects;
+import javax.persistence.Entity;
+import javax.persistence.GeneratedValue;
+import javax.persistence.GenerationType;
+import javax.persistence.Id;
+import javax.persistence.Table;
+
+@Entity
+@Table(name = "tcc_global_tx_event")
+public class GlobalTxEvent {
+
+ @Id
+ @GeneratedValue(strategy = GenerationType.IDENTITY)
+ private Long id;
+ private String globalTxId;
+ private String localTxId;
+ private String parentTxId;
+ private String serviceName;
+ private String instanceId;
+ private String txType;
+ private String status;
+ private Date creationTime;
+ private Date lastModified;
+
+ private GlobalTxEvent() {
+ }
+
+ public GlobalTxEvent(String globalTxId, String localTxId, String parentTxId,
String serviceName,
+ String instanceId, String txType, String status) {
+ this.globalTxId = globalTxId;
+ this.localTxId = localTxId;
+ this.parentTxId = parentTxId;
+ this.serviceName = serviceName;
+ this.instanceId = instanceId;
+ this.txType = txType;
+ this.status = status;
+ this.creationTime = new Date();
+ this.lastModified = new Date();
+ }
+
+ public Long getId() {
+ return id;
+ }
+
+ public String getGlobalTxId() {
+ return globalTxId;
+ }
+
+ public String getLocalTxId() {
+ return localTxId;
+ }
+
+ public String getParentTxId() {
+ return parentTxId;
+ }
+
+ public String getServiceName() {
+ return serviceName;
+ }
+
+ public String getInstanceId() {
+ return instanceId;
+ }
+
+ public String getTxType() {
+ return txType;
+ }
+
+ public String getStatus() {
+ return status;
+ }
+
+ public Date getCreationTime() {
+ return creationTime;
+ }
+
+ public Date getLastModified() {
+ return lastModified;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ GlobalTxEvent that = (GlobalTxEvent) o;
+ return Objects.equals(globalTxId, that.globalTxId) &&
+ Objects.equals(localTxId, that.localTxId) &&
+ Objects.equals(parentTxId, that.parentTxId) &&
+ Objects.equals(serviceName, that.serviceName) &&
+ Objects.equals(instanceId, that.instanceId) &&
+ Objects.equals(txType, that.txType) &&
+ Objects.equals(status, that.status);
+ }
+
+ @Override
+ public int hashCode() {
+
+ return Objects.hash(globalTxId, localTxId, parentTxId, serviceName,
instanceId, txType, status);
+ }
+}
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/GlobalTxEventHistory.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/GlobalTxEventHistory.java
new file mode 100644
index 00000000..f0179ff1
--- /dev/null
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/GlobalTxEventHistory.java
@@ -0,0 +1,93 @@
+/*
+ * 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.alpha.server.tcc.jpa;
+
+import java.util.Date;
+import javax.persistence.Entity;
+import javax.persistence.GeneratedValue;
+import javax.persistence.GenerationType;
+import javax.persistence.Id;
+import javax.persistence.Table;
+
+@Entity
+@Table(name = "tcc_global_tx_event_history")
+public class GlobalTxEventHistory {
+
+ @Id
+ @GeneratedValue(strategy = GenerationType.IDENTITY)
+ private Long id;
+ private String globalTxId;
+ private String localTxId;
+ private String parentTxId;
+ private String serviceName;
+ private String instanceId;
+ private String txType;
+ private Date creationTime;
+ private Date lastModified;
+
+ private GlobalTxEventHistory() {
+ }
+
+ public GlobalTxEventHistory(String globalTxId, String localTxId, String
parentTxId, String serviceName,
+ String instanceId, String txType, Date creationTime, Date lastModified) {
+ this.globalTxId = globalTxId;
+ this.localTxId = localTxId;
+ this.parentTxId = parentTxId;
+ this.serviceName = serviceName;
+ this.instanceId = instanceId;
+ this.txType = txType;
+ this.creationTime = creationTime;
+ this.lastModified = lastModified;
+ }
+
+ public Long getId() {
+ return id;
+ }
+
+ public String getGlobalTxId() {
+ return globalTxId;
+ }
+
+ public String getLocalTxId() {
+ return localTxId;
+ }
+
+ public String getParentTxId() {
+ return parentTxId;
+ }
+
+ public String getServiceName() {
+ return serviceName;
+ }
+
+ public String getInstanceId() {
+ return instanceId;
+ }
+
+ public String getTxType() {
+ return txType;
+ }
+
+ public Date getCreationTime() {
+ return creationTime;
+ }
+
+ public Date getLastModified() {
+ return lastModified;
+ }
+}
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/FinishedEventRepository.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/GlobalTxEventHistoryRepository.java
similarity index 90%
rename from
alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/FinishedEventRepository.java
rename to
alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/GlobalTxEventHistoryRepository.java
index f4370765..a0affa3f 100644
---
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/FinishedEventRepository.java
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/GlobalTxEventHistoryRepository.java
@@ -19,5 +19,6 @@
import org.springframework.data.repository.CrudRepository;
-public interface FinishedEventRepository extends CrudRepository<FinishedEvent,
Long> {
+public interface GlobalTxEventHistoryRepository extends
CrudRepository<GlobalTxEventHistory, Long> {
+
}
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/GlobalTxEventRepository.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/GlobalTxEventRepository.java
new file mode 100644
index 00000000..439e0f53
--- /dev/null
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/GlobalTxEventRepository.java
@@ -0,0 +1,33 @@
+/*
+ * 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.alpha.server.tcc.jpa;
+
+import java.util.List;
+import java.util.Optional;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.CrudRepository;
+
+public interface GlobalTxEventRepository extends CrudRepository<GlobalTxEvent,
Long> {
+
+ @Query(value = "SELECT t FROM GlobalTxEvent AS t WHERE t.globalTxId = ?1")
+ Optional<List<GlobalTxEvent>> findByGlobalTxId(String globalTxId);
+
+ @Query(value = "SELECT t FROM GlobalTxEvent AS t WHERE t.globalTxId = ?1 and
t.localTxId = ?2 and t.txType = ?3")
+ Optional<GlobalTxEvent> findByUniqueKey(String globalTxId, String localTxId,
String txType);
+
+}
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEvent.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEvent.java
index ac4a8ea4..f0cf53ba 100644
---
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEvent.java
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEvent.java
@@ -26,7 +26,7 @@
import javax.persistence.Table;
@Entity
-@Table(name = "TccParticipateEvent")
+@Table(name = "tcc_participate_event")
public class ParticipatedEvent {
@Id
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/FinishedEvent.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEventHistory.java
similarity index 88%
rename from
alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/FinishedEvent.java
rename to
alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEventHistory.java
index 17c14f10..5a82a88e 100644
---
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/FinishedEvent.java
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEventHistory.java
@@ -25,8 +25,8 @@
import javax.persistence.Table;
@Entity
-@Table(name = "TccFinishedEvent")
-public class FinishedEvent {
+@Table(name = "tcc_participate_event_history")
+public class ParticipatedEventHistory {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
@@ -42,10 +42,10 @@
private Date creationTime;
private Date lastModified;
- private FinishedEvent() {
+ private ParticipatedEventHistory() {
}
- public FinishedEvent(String globalTxId, String localTxId, String parentTxId,
String serviceName,
+ public ParticipatedEventHistory(String globalTxId, String localTxId, String
parentTxId, String serviceName,
String instanceId, String confirmMethod, String cancelMethod, String
status) {
this.globalTxId = globalTxId;
this.localTxId = localTxId;
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEventFactory.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEventHistoryRepository.java
similarity index 63%
rename from
alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEventFactory.java
rename to
alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEventHistoryRepository.java
index 27d78841..01bce29d 100644
---
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEventFactory.java
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEventHistoryRepository.java
@@ -17,20 +17,7 @@
package org.apache.servicecomb.saga.alpha.server.tcc.jpa;
-import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTccParticipatedEvent;
+import org.springframework.data.repository.CrudRepository;
-public class ParticipatedEventFactory {
-
- public static ParticipatedEvent create(GrpcTccParticipatedEvent request) {
- return new ParticipatedEvent(
- request.getGlobalTxId(),
- request.getLocalTxId(),
- request.getParentTxId(),
- request.getServiceName(),
- request.getInstanceId(),
- request.getConfirmMethod(),
- request.getCancelMethod(),
- request.getStatus()
- );
- }
+public interface ParticipatedEventHistoryRepository extends
CrudRepository<ParticipatedEventHistory, Long> {
}
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEventRepository.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEventRepository.java
index 35a446f3..b43aaa63 100644
---
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEventRepository.java
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/ParticipatedEventRepository.java
@@ -28,5 +28,5 @@
Optional<List<ParticipatedEvent>> findByGlobalTxId(String globalTxId);
@Query(value = "SELECT t FROM ParticipatedEvent AS t WHERE t.globalTxId = ?1
and t.localTxId = ?2")
- Optional<ParticipatedEvent> findByGlobalTxIdAndLocalTxId(String globalTxId,
String localTxId);
+ Optional<ParticipatedEvent> findByUniqueKey(String globalTxId, String
localTxId);
}
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TccTxType.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TccTxType.java
new file mode 100644
index 00000000..55369ebe
--- /dev/null
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TccTxType.java
@@ -0,0 +1,22 @@
+/*
+ * 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.alpha.server.tcc.jpa;
+
+public enum TccTxType {
+ TCC_START, TCC_END, TCC_END_TIMEOUT
+}
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TxEventFactory.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TxEventFactory.java
new file mode 100644
index 00000000..eec45874
--- /dev/null
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/jpa/TxEventFactory.java
@@ -0,0 +1,63 @@
+/*
+ * 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.alpha.server.tcc.jpa;
+
+import org.apache.servicecomb.saga.common.TransactionStatus;
+import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTccParticipatedEvent;
+import
org.apache.servicecomb.saga.pack.contract.grpc.GrpcTccTransactionEndedEvent;
+import
org.apache.servicecomb.saga.pack.contract.grpc.GrpcTccTransactionStartedEvent;
+
+public class TxEventFactory {
+
+ public static ParticipatedEvent create(GrpcTccParticipatedEvent request) {
+ return new ParticipatedEvent(
+ request.getGlobalTxId(),
+ request.getLocalTxId(),
+ request.getParentTxId(),
+ request.getServiceName(),
+ request.getInstanceId(),
+ request.getConfirmMethod(),
+ request.getCancelMethod(),
+ request.getStatus()
+ );
+ }
+
+ public static GlobalTxEvent create(GrpcTccTransactionStartedEvent request) {
+ return new GlobalTxEvent(
+ request.getGlobalTxId(),
+ request.getLocalTxId(),
+ request.getParentTxId(),
+ request.getServiceName(),
+ request.getInstanceId(),
+ TccTxType.TCC_START.name(),
+ TransactionStatus.Succeed.name()
+ );
+ }
+
+ public static GlobalTxEvent create(GrpcTccTransactionEndedEvent request) {
+ return new GlobalTxEvent(
+ request.getGlobalTxId(),
+ request.getLocalTxId(),
+ request.getParentTxId(),
+ request.getServiceName(),
+ request.getInstanceId(),
+ TccTxType.TCC_END.name(),
+ request.getStatus()
+ );
+ }
+}
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/GlobalTxEventService.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/GlobalTxEventService.java
new file mode 100644
index 00000000..2d4444c4
--- /dev/null
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/GlobalTxEventService.java
@@ -0,0 +1,31 @@
+/*
+ * 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.alpha.server.tcc.service;
+
+import java.util.List;
+import java.util.Optional;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEvent;
+
+public interface GlobalTxEventService {
+
+ boolean addEvent(GlobalTxEvent globalTxEvent);
+
+ Optional<List<GlobalTxEvent>> getEventByGlobalTxId(String globalTxId);
+
+ void migration(String globalTxId, String localTxId);
+}
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/GlobalTxEventServiceImpl.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/GlobalTxEventServiceImpl.java
new file mode 100644
index 00000000..f7631197
--- /dev/null
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/GlobalTxEventServiceImpl.java
@@ -0,0 +1,82 @@
+/*
+ * 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.alpha.server.tcc.service;
+
+import java.lang.invoke.MethodHandles;
+import java.util.List;
+import java.util.Optional;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEventHistory;
+import
org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEventHistoryRepository;
+import
org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEventRepository;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+@Component
+public class GlobalTxEventServiceImpl implements GlobalTxEventService {
+
+ @Autowired
+ private GlobalTxEventRepository hotRepository;
+
+ @Autowired
+ private GlobalTxEventHistoryRepository coldRepository;
+
+ private static final Logger LOG =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ @Override
+ public boolean addEvent(GlobalTxEvent event) {
+ try {
+ if (!hotRepository.findByUniqueKey(event.getGlobalTxId(),
+ event.getLocalTxId(), event.getTxType()).isPresent()) {
+ hotRepository.save(event);
+ }
+ } catch (Exception ex) {
+ LOG.warn("Add globalTxEvent triggered exception, globalTxId:{},
localTxId:{}, txType:{}, ",
+ event.getGlobalTxId(), event.getLocalTxId(), event.getTxType(), ex);
+ return false;
+ }
+ return true;
+ }
+
+ @Override
+ public Optional<List<GlobalTxEvent>> getEventByGlobalTxId(String globalTxId)
{
+ return hotRepository.findByGlobalTxId(globalTxId);
+ }
+
+ @Override
+ public void migration(String globalTxId, String localTxId) {
+ hotRepository.findByGlobalTxId(globalTxId).ifPresent(list ->
+ list.forEach((e) -> {
+ hotRepository.delete(e.getId());
+ GlobalTxEventHistory finishedEvent = new GlobalTxEventHistory(
+ e.getGlobalTxId(),
+ e.getLocalTxId(),
+ e.getParentTxId(),
+ e.getServiceName(),
+ e.getInstanceId(),
+ e.getTxType(),
+ e.getCreationTime(),
+ e.getLastModified()
+ );
+ coldRepository.save(finishedEvent);
+ })
+ );
+ }
+}
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/MemoryEventRegistry.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/MemoryEventRegistry.java
new file mode 100644
index 00000000..615db504
--- /dev/null
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/MemoryEventRegistry.java
@@ -0,0 +1,79 @@
+/*
+ * 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.alpha.server.tcc.service;
+
+import com.google.common.collect.Lists;
+import java.lang.invoke.MethodHandles;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MemoryEventRegistry {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private final static Map<String, Set<GlobalTxEvent>> globalTxMap = new
ConcurrentHashMap<>();
+
+ private final static Map<String, Set<ParticipatedEvent>> participateMap =
new ConcurrentHashMap<>();
+
+ public static void addGlobalTxEvent(GlobalTxEvent globalTxEvent) {
+ globalTxMap
+ .computeIfAbsent(globalTxEvent.getGlobalTxId(), key -> new
LinkedHashSet<>())
+ .add(globalTxEvent);
+ }
+
+ public static void addParticipateEvent(ParticipatedEvent participatedEvent) {
+ participateMap
+ .computeIfAbsent(participatedEvent.getGlobalTxId(), key -> new
LinkedHashSet<>())
+ .add(participatedEvent);
+ }
+
+ public static void migrateParticipate(String globalTxId, String localTxId) {
+ Set<ParticipatedEvent> needRemoveSet =
participateMap.get(globalTxId).stream()
+ .filter((e) -> globalTxId.equals(e.getGlobalTxId()) &&
localTxId.equals(e.getLocalTxId()))
+ .collect(Collectors.toSet());
+ participateMap.get(globalTxId).removeAll(needRemoveSet);
+ }
+
+ public static List<GlobalTxEvent> getGlobalTxEventByGlobalTxId(String
globalTxId) {
+ Set<GlobalTxEvent> events = globalTxMap.get(globalTxId);
+ return null == events ? Lists.newArrayList() : Lists.newArrayList(events);
+ }
+
+ public static List<ParticipatedEvent> getParticipateEventByGlobalTxId(String
globalTxId) {
+ Set<ParticipatedEvent> events = participateMap.get(globalTxId);
+ return null == events ? Lists.newArrayList() : Lists.newArrayList(events);
+ }
+
+ public static void migrationGlobalTxEvent(String globalTxId, String
localTxId) {
+ Set<GlobalTxEvent> needRemoveSet = globalTxMap.get(globalTxId).stream()
+ .filter((e) -> globalTxId.equals(e.getGlobalTxId()) &&
localTxId.equals(e.getLocalTxId()))
+ .collect(Collectors.toSet());
+ globalTxMap.get(globalTxId).removeAll(needRemoveSet);
+ }
+
+
+
+}
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/TransactionEventService.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/ParticipateEventService.java
similarity index 81%
rename from
alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/TransactionEventService.java
rename to
alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/ParticipateEventService.java
index 23c84c49..6d587108 100644
---
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/TransactionEventService.java
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/ParticipateEventService.java
@@ -15,16 +15,17 @@
* limitations under the License.
*/
-package org.apache.servicecomb.saga.alpha.server.tcc;
+package org.apache.servicecomb.saga.alpha.server.tcc.service;
-import java.util.Set;
+import java.util.List;
+import java.util.Optional;
import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
-public interface TransactionEventService {
+public interface ParticipateEventService {
boolean addEvent(ParticipatedEvent participateEvent);
- Set<ParticipatedEvent> getEventByGlobalTxId(String globalTxId);
+ Optional<List<ParticipatedEvent>> getEventByGlobalTxId(String globalTxId);
void migration(String globalTxId, String localTxId);
diff --git
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/RdbTransactionEventService.java
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/ParticipateEventServiceImpl.java
similarity index 62%
rename from
alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/RdbTransactionEventService.java
rename to
alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/ParticipateEventServiceImpl.java
index 626a14a4..ee712145 100644
---
a/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/RdbTransactionEventService.java
+++
b/alpha/alpha-server/src/main/java/org/apache/servicecomb/saga/alpha/server/tcc/service/ParticipateEventServiceImpl.java
@@ -15,60 +15,59 @@
* limitations under the License.
*/
-package org.apache.servicecomb.saga.alpha.server.tcc;
+package org.apache.servicecomb.saga.alpha.server.tcc.service;
-import com.google.common.collect.Sets;
import java.lang.invoke.MethodHandles;
import java.util.List;
import java.util.Optional;
-import java.util.Set;
-import org.apache.servicecomb.saga.alpha.server.tcc.jpa.FinishedEvent;
-import
org.apache.servicecomb.saga.alpha.server.tcc.jpa.FinishedEventRepository;
-import
org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEventRepository;
import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
+import
org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEventHistory;
+import
org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEventHistoryRepository;
+import
org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEventRepository;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
-@Component("rdbTransactionEventService")
-public class RdbTransactionEventService implements TransactionEventService {
+@Component
+public class ParticipateEventServiceImpl implements ParticipateEventService {
@Autowired
- private ParticipatedEventRepository participatedEventRepository;
+ private ParticipatedEventRepository hotRepository;
@Autowired
- private FinishedEventRepository finishedEventRepository;
+ private ParticipatedEventHistoryRepository coldRepository;
private static final Logger LOG =
LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
@Override
+ @Transactional
public boolean addEvent(ParticipatedEvent event) {
try {
- if (!participatedEventRepository.findByGlobalTxIdAndLocalTxId(
+ if (!hotRepository.findByUniqueKey(
event.getGlobalTxId(), event.getLocalTxId()).isPresent()) {
- participatedEventRepository.save(event);
+ hotRepository.save(event);
}
} catch (Exception ex) {
- LOG.warn("add event triggered exception: ", ex);
+ LOG.warn("Add participateEvent triggered exception, globalTxId:{},
localTxId:{}, ",
+ event.getGlobalTxId(), event.getLocalTxId(), ex);
return false;
}
return true;
}
@Override
- public Set<ParticipatedEvent> getEventByGlobalTxId(String globalTxId) {
- Optional<List<ParticipatedEvent>> list =
participatedEventRepository.findByGlobalTxId(globalTxId);
- return list.map(Sets::newHashSet).orElseGet(Sets::newHashSet);
+ public Optional<List<ParticipatedEvent>> getEventByGlobalTxId(String
globalTxId) {
+ return hotRepository.findByGlobalTxId(globalTxId);
}
@Override
@Transactional
public void migration(String globalTxId, String localTxId) {
- participatedEventRepository.findByGlobalTxIdAndLocalTxId(globalTxId,
localTxId).ifPresent( e -> {
- participatedEventRepository.delete(e.getId());
- FinishedEvent finishedEvent = new FinishedEvent(
+ hotRepository.findByUniqueKey(globalTxId, localTxId).ifPresent( e -> {
+ hotRepository.delete(e.getId());
+ ParticipatedEventHistory finishedEvent = new ParticipatedEventHistory(
e.getGlobalTxId(),
e.getLocalTxId(),
e.getParentTxId(),
@@ -78,7 +77,7 @@ public void migration(String globalTxId, String localTxId) {
e.getCancelMethod(),
e.getStatus()
);
- finishedEventRepository.save(finishedEvent);
+ coldRepository.save(finishedEvent);
});
}
}
diff --git a/alpha/alpha-server/src/main/resources/schema-mysql.sql
b/alpha/alpha-server/src/main/resources/schema-mysql.sql
index d28a16b2..862c0434 100644
--- a/alpha/alpha-server/src/main/resources/schema-mysql.sql
+++ b/alpha/alpha-server/src/main/resources/schema-mysql.sql
@@ -67,7 +67,36 @@ CREATE TABLE IF NOT EXISTS TxTimeout (
INDEX saga_timeouts_index (surrogateId, expiryTime, globalTxId, localTxId,
status)
) DEFAULT CHARSET=utf8;
-CREATE TABLE IF NOT EXISTS TccParticipateEvent (
+CREATE TABLE IF NOT EXISTS tcc_global_tx_event (
+ id bigint NOT NULL AUTO_INCREMENT,
+ globalTxId varchar(36) NOT NULL,
+ localTxId varchar(36) NOT NULL,
+ parentTxId varchar(36) DEFAULT NULL,
+ serviceName varchar(36) NOT NULL,
+ instanceId varchar(36) NOT NULL,
+ txType varchar(12),
+ creationTime datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ lastModified datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ PRIMARY KEY (id),
+ UNIQUE INDEX tcc_global_tx_event_index (globalTxId, localTxId, parentTxId,
txType)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8;
+
+CREATE TABLE IF NOT EXISTS tcc_global_tx_event_history (
+ id bigint NOT NULL AUTO_INCREMENT,
+ globalTxId varchar(36) NOT NULL,
+ localTxId varchar(36) NOT NULL,
+ parentTxId varchar(36) DEFAULT NULL,
+ serviceName varchar(36) NOT NULL,
+ instanceId varchar(36) NOT NULL,
+ txType varchar(12),
+ status varchar(12),
+ creationTime datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ lastModified datetime NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ PRIMARY KEY (id),
+ UNIQUE INDEX tcc_global_tx_event_history_index (globalTxId, localTxId,
parentTxId, txType)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8;
+
+CREATE TABLE IF NOT EXISTS tcc_participate_event (
id bigint NOT NULL AUTO_INCREMENT,
serviceName varchar(36) NOT NULL,
instanceId varchar(36) NOT NULL,
@@ -83,7 +112,7 @@ CREATE TABLE IF NOT EXISTS TccParticipateEvent (
UNIQUE INDEX tcc_participate_event_index (globalTxId, localTxId, parentTxId)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;
-CREATE TABLE IF NOT EXISTS TccFinishedEvent (
+CREATE TABLE IF NOT EXISTS tcc_participate_event_history (
id bigint NOT NULL AUTO_INCREMENT,
serviceName varchar(36) NOT NULL,
instanceId varchar(36) NOT NULL,
diff --git a/alpha/alpha-server/src/main/resources/schema-postgresql.sql
b/alpha/alpha-server/src/main/resources/schema-postgresql.sql
index d55b7d8c..5f81e477 100644
--- a/alpha/alpha-server/src/main/resources/schema-postgresql.sql
+++ b/alpha/alpha-server/src/main/resources/schema-postgresql.sql
@@ -69,7 +69,34 @@ CREATE TABLE IF NOT EXISTS TxTimeout (
CREATE INDEX IF NOT EXISTS saga_timeouts_index ON TxTimeout (surrogateId,
expiryTime, globalTxId, localTxId, status);
-CREATE TABLE IF NOT EXISTS TccParticipateEvent (
+CREATE TABLE IF NOT EXISTS tcc_global_tx_event (
+ id BIGSERIAL PRIMARY KEY,
+ globalTxId varchar(36) NOT NULL,
+ localTxId varchar(36) NOT NULL,
+ parentTxId varchar(36) DEFAULT NULL,
+ serviceName varchar(36) NOT NULL,
+ instanceId varchar(36) NOT NULL,
+ txType varchar(12),
+ creationTime timestamp(6) NOT NULL DEFAULT CURRENT_DATE,
+ lastModified timestamp(6) NOT NULL DEFAULT CURRENT_DATE
+)
+CREATE UNIQUE INDEX IF NOT EXISTS tcc_global_tx_event_index ON
tcc_global_tx_event (globalTxId, localTxId, parentTxId, txType);
+
+CREATE TABLE IF NOT EXISTS tcc_global_tx_event_history (
+ id BIGSERIAL PRIMARY KEY,
+ globalTxId varchar(36) NOT NULL,
+ localTxId varchar(36) NOT NULL,
+ parentTxId varchar(36) DEFAULT NULL,
+ serviceName varchar(36) NOT NULL,
+ instanceId varchar(36) NOT NULL,
+ txType varchar(12),
+ status varchar(12),
+ creationTime timestamp(6) NOT NULL DEFAULT CURRENT_DATE,
+ lastModified timestamp(6) NOT NULL DEFAULT CURRENT_DATE
+)
+CREATE UNIQUE INDEX IF NOT EXISTS tcc_global_tx_event_history_index ON
tcc_global_tx_event_history (globalTxId, localTxId, parentTxId, txType);
+
+CREATE TABLE IF NOT EXISTS tcc_participate_event (
id BIGSERIAL PRIMARY KEY,
serviceName varchar(36) NOT NULL,
instanceId varchar(36) NOT NULL,
@@ -83,9 +110,9 @@ CREATE TABLE IF NOT EXISTS TccParticipateEvent (
lastModified timestamp(6) NOT NULL DEFAULT CURRENT_DATE
);
-CREATE UNIQUE INDEX IF NOT EXISTS tcc_participate_event_index ON
TccParticipateEvent (globalTxId, localTxId, parentTxId);
+CREATE UNIQUE INDEX IF NOT EXISTS tcc_participate_event_index ON
tcc_participate_event (globalTxId, localTxId, parentTxId);
-CREATE TABLE IF NOT EXISTS TccFinishedEvent (
+CREATE TABLE IF NOT EXISTS tcc_participate_event_history (
id BIGSERIAL PRIMARY KEY,
serviceName varchar(36) NOT NULL,
instanceId varchar(36) NOT NULL,
@@ -99,4 +126,4 @@ CREATE TABLE IF NOT EXISTS TccFinishedEvent (
lastModified timestamp(6) NOT NULL DEFAULT CURRENT_DATE
);
-CREATE UNIQUE INDEX IF NOT EXISTS tcc_finished_event_index ON TccFinishedEvent
(globalTxId, localTxId, parentTxId);
+CREATE UNIQUE INDEX IF NOT EXISTS tcc_finished_event_index ON
tcc_participate_event_history (globalTxId, localTxId, parentTxId);
diff --git
a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/AlphaTccServerTest.java
b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/AlphaTccServerTest.java
index 0174d13e..43c32592 100644
---
a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/AlphaTccServerTest.java
+++
b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/AlphaTccServerTest.java
@@ -26,14 +26,18 @@
import io.grpc.ManagedChannel;
import io.grpc.netty.NettyChannelBuilder;
+import java.util.Iterator;
+import java.util.List;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.ConcurrentLinkedQueue;
-import org.apache.servicecomb.saga.alpha.server.AlphaApplication;
+import org.apache.servicecomb.saga.alpha.server.tcc.TccTxEventFacade;
import
org.apache.servicecomb.saga.alpha.server.tcc.callback.GrpcOmegaTccCallback;
-import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
import
org.apache.servicecomb.saga.alpha.server.tcc.callback.OmegaCallbacksRegistry;
-import org.apache.servicecomb.saga.alpha.server.tcc.TransactionEventService;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.GlobalTxEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.ParticipatedEvent;
+import org.apache.servicecomb.saga.alpha.server.tcc.jpa.TccTxType;
+import org.apache.servicecomb.saga.common.TransactionStatus;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcAck;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcServiceConfig;
import org.apache.servicecomb.saga.pack.contract.grpc.GrpcTccCoordinateCommand;
@@ -48,20 +52,8 @@
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.test.context.SpringBootTest;
-import org.springframework.test.context.junit4.SpringRunner;
-
-@RunWith(SpringRunner.class)
-@SpringBootTest(classes = {AlphaApplication.class},
- properties = {
- "alpha.server.host=0.0.0.0",
- "alpha.server.port=8090"
- })
-public class AlphaTccServerTest {
- private static final int port = 8090;
+public abstract class AlphaTccServerTest {
protected static ManagedChannel clientChannel;
@@ -80,7 +72,6 @@
private final String confirmMethod = "confirm";
private final String cancelMethod = "cancel";
-
private final String serviceName = uniquify("serviceName");
private final String instanceId = uniquify("instanceId");
@@ -89,13 +80,7 @@
.setInstanceId(instanceId)
.build();
- @Autowired
- private TransactionEventService transactionEventService;
-
- @BeforeClass
- public static void setupClientChannel() {
- clientChannel = NettyChannelBuilder.forAddress("localhost",
port).usePlaintext().build();
- }
+ public abstract TccTxEventFacade getTccTxEventFacade();
@AfterClass
public static void tearDown() {
@@ -108,7 +93,6 @@ public void after() {
blockingStub.onDisconnected(serviceConfig);
}
- @Test
public void assertOnConnect() {
asyncStub.onConnected(serviceConfig, commandStreamObserver);
awaitUntilConnected();
@@ -117,7 +101,6 @@ public void assertOnConnect() {
);
}
- @Test
public void assertOnDisConnect() {
asyncStub.onConnected(serviceConfig, commandStreamObserver);
awaitUntilConnected();
@@ -132,14 +115,41 @@ private void awaitUntilConnected() {
await().atMost(2, SECONDS).until(() -> null !=
(OmegaCallbacksRegistry.getRegistry().get(serviceName)));
}
- @Test
+ public void assertOnTransactionStart() {
+ asyncStub.onConnected(serviceConfig, commandStreamObserver);
+ awaitUntilConnected();
+ blockingStub.onTccTransactionStarted(newTxStart());
+ blockingStub.onTccTransactionStarted(newTxStart());
+ blockingStub.onTccTransactionEnded(newTxEnd("Succeed"));
+ List<GlobalTxEvent> events =
getTccTxEventFacade().getGlobalTxEventByGlobalTxId(globalTxId);
+ assertThat(events.size(), is(2));
+
+ Iterator<GlobalTxEvent> iterator = events.iterator();
+ GlobalTxEvent event = iterator.next();
+ assertThat(event.getGlobalTxId(), is(globalTxId));
+ assertThat(event.getLocalTxId(), is(localTxId));
+ assertThat(event.getInstanceId(), is(instanceId));
+ assertThat(event.getServiceName(), is(serviceName));
+ assertThat(event.getTxType(), is(TccTxType.TCC_START.name()));
+ assertThat(event.getStatus(), is(TransactionStatus.Succeed.name()));
+
+ event = iterator.next();
+ assertThat(event.getGlobalTxId(), is(globalTxId));
+ assertThat(event.getLocalTxId(), is(localTxId));
+ assertThat(event.getInstanceId(), is(instanceId));
+ assertThat(event.getServiceName(), is(serviceName));
+ assertThat(event.getTxType(), is(TccTxType.TCC_END.name()));
+ assertThat(event.getStatus(), is(TransactionStatus.Succeed.name()));
+ }
+
public void assertOnParticipated() {
asyncStub.onConnected(serviceConfig, commandStreamObserver);
awaitUntilConnected();
blockingStub.participate(newParticipatedEvent("Succeed"));
blockingStub.participate(newParticipatedEvent("Succeed"));
-
assertThat(transactionEventService.getEventByGlobalTxId(globalTxId).size(),
is(1));
- ParticipatedEvent event =
transactionEventService.getEventByGlobalTxId(globalTxId).iterator().next();
+ List<ParticipatedEvent> events =
getTccTxEventFacade().getParticipateEventByGlobalTxId(globalTxId);
+ assertThat(events.size(), is(1));
+ ParticipatedEvent event = events.iterator().next();
assertThat(event.getGlobalTxId(), is(globalTxId));
assertThat(event.getLocalTxId(), is(localTxId));
assertThat(event.getInstanceId(), is(instanceId));
@@ -168,7 +178,6 @@ public void assertOnTccTransactionSucceedEnded() {
assertThat(result.getAborted(), is(false));
}
- @Test
public void assertOnTccTransactionFailedEnded() {
asyncStub.onConnected(serviceConfig, commandStreamObserver);
awaitUntilConnected();
@@ -185,7 +194,6 @@ public void assertOnTccTransactionFailedEnded() {
assertThat(commandStreamObserver.isCompleted(), is(false));
}
- @Test
public void assertOnCallbackNotExist() {
asyncStub.onConnected(serviceConfig, commandStreamObserver);
awaitUntilConnected();
@@ -197,7 +205,6 @@ public void assertOnCallbackNotExist() {
assertThat(result.getAborted(), is(true));
}
- @Test
public void assertOnCallbacksExecuteError() {
asyncStub.onConnected(serviceConfig, commandStreamObserver);
awaitUntilConnected();
@@ -211,7 +218,6 @@ public void assertOnCallbacksExecuteError() {
assertThat(OmegaCallbacksRegistry.getRegistry().get(serviceName).size(),
is(0));
}
- @Test
public void assertOnSwitchOtherCallbackInstance() {
asyncStub.onConnected(serviceConfig, commandStreamObserver);
GrpcServiceConfig config = GrpcServiceConfig.newBuilder()
@@ -254,6 +260,9 @@ private GrpcTccTransactionStartedEvent newTxStart() {
return GrpcTccTransactionStartedEvent.newBuilder()
.setGlobalTxId(globalTxId)
.setLocalTxId(localTxId)
+ .setServiceName(serviceName)
+ .setInstanceId(instanceId)
+ .setLocalTxId(localTxId)
.build();
}
@@ -261,6 +270,9 @@ private GrpcTccTransactionEndedEvent newTxEnd(String
status) {
return GrpcTccTransactionEndedEvent.newBuilder()
.setGlobalTxId(globalTxId)
.setLocalTxId(localTxId)
+ .setServiceName(serviceName)
+ .setInstanceId(instanceId)
+ .setLocalTxId(localTxId)
.setStatus(status)
.build();
}
@@ -279,5 +291,4 @@ private GrpcTccCoordinatedEvent newCoordinatedEvent(String
status, String method
private GrpcAck onReceivedCoordinateCommand(GrpcTccCoordinateCommand
command) {
return GrpcAck.newBuilder().setAborted(false).build();
}
-
}
diff --git
a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/MemoryAlphaTccServerTest.java
b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/MemoryAlphaTccServerTest.java
new file mode 100644
index 00000000..59d3c193
--- /dev/null
+++
b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/MemoryAlphaTccServerTest.java
@@ -0,0 +1,93 @@
+/*
+ * 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.alpha.tcc.server;
+
+import io.grpc.netty.NettyChannelBuilder;
+import org.apache.servicecomb.saga.alpha.server.AlphaApplication;
+import org.apache.servicecomb.saga.alpha.server.tcc.TccTxEventFacade;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.test.context.junit4.SpringRunner;
+
+@RunWith(SpringRunner.class)
+@SpringBootTest(classes = {AlphaApplication.class},
+ properties = {
+ "alpha.server.host=0.0.0.0",
+ "alpha.server.port=8090",
+ "alpha.server.storage=local"
+ })
+public class MemoryAlphaTccServerTest extends AlphaTccServerTest {
+
+ @BeforeClass
+ public static void setupClientChannel() {
+ clientChannel = NettyChannelBuilder.forAddress("localhost",
8090).usePlaintext().build();
+ }
+
+ @Autowired
+ @Qualifier("defaultTccTxEventFacade")
+ private TccTxEventFacade tccTxEventFacade;
+
+ @Override
+ public TccTxEventFacade getTccTxEventFacade() {
+ return tccTxEventFacade;
+ }
+
+ @Test
+ public void assertOnDisConnect() {
+ super.assertOnDisConnect();
+ }
+
+ @Test
+ public void assertOnTransactionStart() {
+ super.assertOnTransactionStart();
+ }
+
+ @Test
+ public void assertOnParticipated() {
+ super.assertOnParticipated();
+ }
+
+ @Test
+ public void assertOnTccTransactionSucceedEnded() {
+ super.assertOnTccTransactionSucceedEnded();
+ }
+
+ @Test
+ public void assertOnTccTransactionFailedEnded() {
+ super.assertOnTccTransactionFailedEnded();
+ }
+
+ @Test
+ public void assertOnCallbackNotExist() {
+ super.assertOnCallbackNotExist();
+ }
+
+ @Test
+ public void assertOnCallbacksExecuteError() {
+ super.assertOnCallbacksExecuteError();
+ }
+
+ @Test
+ public void assertOnSwitchOtherCallbackInstance() {
+ super.assertOnSwitchOtherCallbackInstance();
+ }
+}
diff --git
a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/RdbAlphaTccServerTest.java
b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/RdbAlphaTccServerTest.java
index 25f1b8d8..e2db6ebb 100644
---
a/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/RdbAlphaTccServerTest.java
+++
b/alpha/alpha-server/src/test/java/org/apache/servicecomb/saga/alpha/tcc/server/RdbAlphaTccServerTest.java
@@ -17,8 +17,14 @@
package org.apache.servicecomb.saga.alpha.tcc.server;
+import io.grpc.netty.NettyChannelBuilder;
import org.apache.servicecomb.saga.alpha.server.AlphaApplication;
+import org.apache.servicecomb.saga.alpha.server.tcc.TccTxEventFacade;
+import org.junit.BeforeClass;
+import org.junit.Test;
import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
@@ -26,8 +32,62 @@
@SpringBootTest(classes = {AlphaApplication.class},
properties = {
"alpha.server.host=0.0.0.0",
- "alpha.server.port=8090",
+ "alpha.server.port=8091",
"alpha.server.storage=rdb"
})
-public class RdbAlphaTccServerTest extends AlphaTccServerTest{
+public class RdbAlphaTccServerTest extends AlphaTccServerTest {
+
+ @BeforeClass
+ public static void setupClientChannel() {
+ clientChannel = NettyChannelBuilder.forAddress("localhost",
8091).usePlaintext().build();
+ }
+
+ @Autowired
+ @Qualifier("rdbTccTxEventFacade")
+ private TccTxEventFacade tccTxEventFacade;
+
+ @Override
+ public TccTxEventFacade getTccTxEventFacade() {
+ return tccTxEventFacade;
+ }
+
+ @Test
+ public void assertOnDisConnect() {
+ super.assertOnDisConnect();
+ }
+
+ @Test
+ public void assertOnTransactionStart() {
+ super.assertOnTransactionStart();
+ }
+
+ @Test
+ public void assertOnParticipated() {
+ super.assertOnParticipated();
+ }
+
+ @Test
+ public void assertOnTccTransactionSucceedEnded() {
+ super.assertOnTccTransactionSucceedEnded();
+ }
+
+ @Test
+ public void assertOnTccTransactionFailedEnded() {
+ super.assertOnTccTransactionFailedEnded();
+ }
+
+ @Test
+ public void assertOnCallbackNotExist() {
+ super.assertOnCallbackNotExist();
+ }
+
+ @Test
+ public void assertOnCallbacksExecuteError() {
+ super.assertOnCallbacksExecuteError();
+ }
+
+ @Test
+ public void assertOnSwitchOtherCallbackInstance() {
+ super.assertOnSwitchOtherCallbackInstance();
+ }
}
diff --git a/alpha/alpha-server/src/test/resources/schema.sql
b/alpha/alpha-server/src/test/resources/schema.sql
index b464de1d..b184b65b 100644
--- a/alpha/alpha-server/src/test/resources/schema.sql
+++ b/alpha/alpha-server/src/test/resources/schema.sql
@@ -60,7 +60,32 @@ CREATE TABLE IF NOT EXISTS TxTimeout (
version bigint NOT NULL
);
-CREATE TABLE IF NOT EXISTS TccParticipateEvent (
+CREATE TABLE IF NOT EXISTS tcc_global_tx_event (
+ id bigint GENERATED BY DEFAULT AS IDENTITY(START WITH 1, INCREMENT BY 1)
PRIMARY KEY,
+ globalTxId varchar(36) NOT NULL,
+ localTxId varchar(36) NOT NULL,
+ parentTxId varchar(36) DEFAULT NULL,
+ serviceName varchar(36) NOT NULL,
+ instanceId varchar(36) NOT NULL,
+ txType varchar(12),
+ status varchar(12),
+ creationTime TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL,
+ lastModified TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS tcc_global_tx_event_history (
+ id bigint GENERATED BY DEFAULT AS IDENTITY(START WITH 1, INCREMENT BY 1)
PRIMARY KEY,
+ globalTxId varchar(36) NOT NULL,
+ localTxId varchar(36) NOT NULL,
+ parentTxId varchar(36) DEFAULT NULL,
+ serviceName varchar(36) NOT NULL,
+ instanceId varchar(36) NOT NULL,
+ txType varchar(12),
+ creationTime TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL,
+ lastModified TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS tcc_participate_event (
id bigint GENERATED BY DEFAULT AS IDENTITY(START WITH 1, INCREMENT BY 1)
PRIMARY KEY,
serviceName varchar(36) NOT NULL,
instanceId varchar(36) NOT NULL,
@@ -74,7 +99,7 @@ CREATE TABLE IF NOT EXISTS TccParticipateEvent (
lastModified TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL
);
-CREATE TABLE IF NOT EXISTS TccFinishedEvent (
+CREATE TABLE IF NOT EXISTS tcc_participate_event_history (
id bigint GENERATED BY DEFAULT AS IDENTITY(START WITH 1, INCREMENT BY 1)
PRIMARY KEY,
serviceName varchar(36) NOT NULL,
instanceId varchar(36) NOT NULL,
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
> Alpha should pesistend the received TCC events
> -------------------------------------------------
>
> Key: SCB-877
> URL: https://issues.apache.org/jira/browse/SCB-877
> Project: Apache ServiceComb
> Issue Type: Sub-task
> Reporter: Willem Jiang
> Assignee: cherrylzhao
> Priority: Major
> Fix For: saga-0.3.0
>
>
> It's important that we need to keep all the received the TCC events
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
