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 2020/01/09 14:27:00 UTC
[servicecomb-pack] branch master updated (133a2fa -> 197d986)
This is an automated email from the ASF dual-hosted git repository.
ningjiang pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-pack.git.
from 133a2fa Revert "SCB-1695 Add attribute mode to @Compensable annotation"
new cac096b SCB-1707 forward compensation only sends once a failure event if multiple retries fail
new 2259476 SCB-1707 Add prefix 'forward' to parameter retries and timeout
new 197d986 SCB-1707 Update test cases for forward retry failure scenario
The 3 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../pack/alpha/core/fsm/event/TxStartedEvent.java | 14 +++---
.../pack/alpha/fsm/domain/AddTxEventDomain.java | 2 +-
.../pack/alpha/server/GrpcTxEventEndpointImpl.java | 4 +-
.../alpha/server/fsm/GrpcSagaEventService.java | 4 +-
.../pack/alpha/server/AlphaIntegrationTest.java | 6 +--
.../server/AlphaIntegrationWithRandomPortTest.java | 6 +--
.../alpha/server/fsm/OmegaEventSagaSimulator.java | 6 +--
.../ExplicitTransactionContextIT.java | 25 +++++------
.../integration/tests/resttemplate/PackIT.java | 26 +++++------
.../grpc/saga/GrpcSagaClientMessageSender.java | 8 +---
.../grpc/saga/SagaLoadBalancedSenderTestBase.java | 4 +-
.../spring/TransactionInterceptionTest.java | 18 +++-----
.../omega/transaction/AbstractRecoveryPolicy.java | 8 ++--
.../omega/transaction/CompensableInterceptor.java | 4 +-
.../pack/omega/transaction/DefaultRecovery.java | 11 +++--
.../omega/transaction/EventAwareInterceptor.java | 2 +-
.../pack/omega/transaction/ForwardRecovery.java | 10 ++---
.../transaction/NoOpEventAwareInterceptor.java | 2 +-
.../pack/omega/transaction/RecoveryPolicy.java | 2 +-
.../omega/transaction/RecoveryPolicyFactory.java | 4 +-
.../pack/omega/transaction/TransactionAspect.java | 6 +--
.../pack/omega/transaction/TxEvent.java | 12 ++---
.../pack/omega/transaction/TxStartedEvent.java | 4 +-
.../transaction/CompensableInterceptorTest.java | 6 +--
.../omega/transaction/DefaultRecoveryTest.java | 6 +--
.../omega/transaction/ForwardRecoveryTest.java | 7 ++-
.../omega/transaction/TransactionAspectTest.java | 51 ++++++++++++----------
.../src/main/proto/GrpcTxEvent.proto | 4 +-
28 files changed, 125 insertions(+), 137 deletions(-)
[servicecomb-pack] 03/03: SCB-1707 Update test cases for forward
retry failure scenario
Posted by ni...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ningjiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-pack.git
commit 197d9860a1dc040ad3a4922c21e73310e3954da4
Author: Lei Zhang <zh...@apache.org>
AuthorDate: Wed Jan 8 10:14:55 2020 +0800
SCB-1707 Update test cases for forward retry failure scenario
---
.../integration/tests/resttemplate/PackIT.java | 26 +++++------
.../omega/transaction/ForwardRecoveryTest.java | 7 ++-
.../omega/transaction/TransactionAspectTest.java | 51 ++++++++++++----------
3 files changed, 41 insertions(+), 43 deletions(-)
diff --git a/integration-tests/pack-tests/src/test/java/org/apache/servicecomb/pack/integration/tests/resttemplate/PackIT.java b/integration-tests/pack-tests/src/test/java/org/apache/servicecomb/pack/integration/tests/resttemplate/PackIT.java
index 588cf08..246e951 100644
--- a/integration-tests/pack-tests/src/test/java/org/apache/servicecomb/pack/integration/tests/resttemplate/PackIT.java
+++ b/integration-tests/pack-tests/src/test/java/org/apache/servicecomb/pack/integration/tests/resttemplate/PackIT.java
@@ -259,23 +259,22 @@ public class PackIT {
assertThat(entity.getStatusCode(), is(OK));
assertThat(entity.getBody(), is("Greetings, eric; Welcome to visit the zoo, eric"));
- await().atMost(10, SECONDS).until(() -> eventRepo.count() == 8);
+ await().atMost(10, SECONDS).until(() -> eventRepo.count() == 7);
List<String> distinctGlobalTxIds = eventRepo.findDistinctGlobalTxId();
assertThat(distinctGlobalTxIds.size(), greaterThanOrEqualTo(1));
String globalTxId = distinctGlobalTxIds.get(0);
List<TxEvent> events = eventRepo.findByGlobalTxIdOrderByCreationTime(globalTxId);
- assertThat(events.size(), is(8));
+ assertThat(events.size(), is(7));
assertThat(events.get(0).type(), is("SagaStartedEvent"));
assertThat(events.get(1).type(), is("TxStartedEvent"));
assertThat(events.get(2).type(), is("TxEndedEvent"));
assertThat(events.get(3).type(), is("TxStartedEvent"));
- assertThat(events.get(4).type(), is("TxAbortedEvent"));
- assertThat(events.get(5).type(), is("TxStartedEvent"));
- assertThat(events.get(6).type(), is("TxEndedEvent"));
- assertThat(events.get(7).type(), is("SagaEndedEvent"));
+ assertThat(events.get(4).type(), is("TxStartedEvent"));
+ assertThat(events.get(5).type(), is("TxEndedEvent"));
+ assertThat(events.get(6).type(), is("SagaEndedEvent"));
assertThat(compensatedMessages.isEmpty(), is(true));
}
@@ -290,27 +289,24 @@ public class PackIT {
assertThat(entity.getStatusCode(), is(INTERNAL_SERVER_ERROR));
- await().atMost(10, SECONDS).until(() -> eventRepo.count() == 12);
+ await().atMost(10, SECONDS).until(() -> eventRepo.count() == 10);
List<String> distinctGlobalTxIds = eventRepo.findDistinctGlobalTxId();
assertThat(distinctGlobalTxIds.size(), greaterThanOrEqualTo(1));
String globalTxId = distinctGlobalTxIds.get(0);
List<TxEvent> events = eventRepo.findByGlobalTxIdOrderByCreationTime(globalTxId);
- assertThat(events.size(), is(12));
+ assertThat(events.size(), is(10));
assertThat(events.get(0).type(), is("SagaStartedEvent"));
assertThat(events.get(1).type(), is("TxStartedEvent"));
assertThat(events.get(2).type(), is("TxEndedEvent"));
assertThat(events.get(3).type(), is("TxStartedEvent"));
- assertThat(events.get(4).type(), is("TxAbortedEvent"));
+ assertThat(events.get(4).type(), is("TxStartedEvent"));
assertThat(events.get(5).type(), is("TxStartedEvent"));
- assertThat(events.get(6).type(), is("TxAbortedEvent"));
- assertThat(events.get(7).type(), is("TxStartedEvent"));
- assertThat(events.get(8).type(), is("TxAbortedEvent"));
- // This event is for the whole saga event
- assertThat(events.get(9).type(), is("TxAbortedEvent"));
- assertThat(events.get(10).type(), is("TxCompensatedEvent"));
+ assertThat(events.get(7).type(), is("TxAbortedEvent"));
+ assertThat(events.get(8).type(), is("TxCompensatedEvent"));
+ assertThat(events.get(9).type(), is("SagaEndedEvent"));
assertThat(compensatedMessages, Matchers.contains("Goodbye, " + GreetingController.TRESPASSER));
}
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/ForwardRecoveryTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/ForwardRecoveryTest.java
index 1d16c93..97f5d66 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/ForwardRecoveryTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/ForwardRecoveryTest.java
@@ -155,11 +155,10 @@ public class ForwardRecoveryTest {
containsString("Failed to handle tx because it is interrupted")));
}
- assertThat(messages.size(), is(4));
+ assertThat(messages.size(), is(3));
assertThat(messages.get(0).type(), is(EventType.TxStartedEvent));
- assertThat(messages.get(1).type(), is(EventType.TxAbortedEvent));
- assertThat(messages.get(2).type(), is(EventType.TxStartedEvent));
- assertThat(messages.get(3).type(), is(EventType.TxAbortedEvent));
+ assertThat(messages.get(1).type(), is(EventType.TxStartedEvent));
+ assertThat(messages.get(2).type(), is(EventType.TxAbortedEvent));
}
@Test
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/TransactionAspectTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/TransactionAspectTest.java
index 39a32c4..9de70a5 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/TransactionAspectTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/TransactionAspectTest.java
@@ -129,7 +129,7 @@ public class TransactionAspectTest {
assertThat(startedEvent.localTxId(), is(newLocalTxId));
assertThat(startedEvent.parentTxId(), is(transactionLocalTxId));
assertThat(startedEvent.type(), is(EventType.TxStartedEvent));
- assertThat(startedEvent.retries(), is(0));
+ assertThat(startedEvent.forwardRetries(), is(0));
assertThat(startedEvent.retryMethod().isEmpty(), is(true));
TxEvent endedEvent = messages.get(1);
@@ -164,7 +164,7 @@ public class TransactionAspectTest {
assertThat(startedEvent.localTxId(), is(newLocalTxId));
assertThat(startedEvent.parentTxId(), is(localTxId));
assertThat(startedEvent.type(), is(EventType.TxStartedEvent));
- assertThat(startedEvent.retries(), is(0));
+ assertThat(startedEvent.forwardRetries(), is(0));
assertThat(startedEvent.retryMethod().isEmpty(), is(true));
TxEvent endedEvent = messages.get(1);
@@ -305,6 +305,13 @@ public class TransactionAspectTest {
}
}
+ /**
+ * Send TxAbortedEvent after three failed retries
+ * TxStartedEvent retry 1
+ * TxStartedEvent retry 2
+ * TxStartedEvent retry 3
+ * TxAbortedEvent
+ * */
@Test
public void retryReachesMaximumAndForwardException() throws Throwable {
RuntimeException oops = new RuntimeException("oops");
@@ -318,32 +325,32 @@ public class TransactionAspectTest {
assertThat(e.getMessage(), is("oops"));
}
- assertThat(messages.size(), is(6));
+ assertThat(messages.size(), is(4));
+ // TxStartedEvent retry 1
TxEvent startedEvent1 = messages.get(0);
assertThat(startedEvent1.globalTxId(), is(globalTxId));
assertThat(startedEvent1.localTxId(), is(newLocalTxId));
assertThat(startedEvent1.parentTxId(), is(localTxId));
assertThat(startedEvent1.type(), is(EventType.TxStartedEvent));
- assertThat(startedEvent1.retries(), is(3));
+ assertThat(startedEvent1.forwardRetries(), is(3));
assertThat(startedEvent1.retryMethod(),
is(this.getClass().getDeclaredMethod("doNothing").toString()));
- assertThat(messages.get(1).type(), is(EventType.TxAbortedEvent));
-
- TxEvent startedEvent2 = messages.get(2);
+ // TxStartedEvent retry 2
+ TxEvent startedEvent2 = messages.get(1);
assertThat(startedEvent2.localTxId(), is(newLocalTxId));
assertThat(startedEvent2.type(), is(EventType.TxStartedEvent));
- assertThat(startedEvent2.retries(), is(2));
-
- assertThat(messages.get(3).type(), is(EventType.TxAbortedEvent));
+ assertThat(startedEvent2.forwardRetries(), is(2));
- TxEvent startedEvent3 = messages.get(4);
+ // TxStartedEvent retry 3
+ TxEvent startedEvent3 = messages.get(2);
assertThat(startedEvent3.localTxId(), is(newLocalTxId));
assertThat(startedEvent3.type(), is(EventType.TxStartedEvent));
- assertThat(startedEvent3.retries(), is(1));
+ assertThat(startedEvent3.forwardRetries(), is(1));
- assertThat(messages.get(5).type(), is(EventType.TxAbortedEvent));
+ // TxAbortedEvent
+ assertThat(messages.get(3).type(), is(EventType.TxAbortedEvent));
assertThat(omegaContext.globalTxId(), is(globalTxId));
assertThat(omegaContext.localTxId(), is(localTxId));
@@ -357,32 +364,28 @@ public class TransactionAspectTest {
aspect.advise(joinPoint, compensable);
- assertThat(messages.size(), is(6));
+ assertThat(messages.size(), is(4));
TxEvent startedEvent1 = messages.get(0);
assertThat(startedEvent1.globalTxId(), is(globalTxId));
assertThat(startedEvent1.localTxId(), is(newLocalTxId));
assertThat(startedEvent1.parentTxId(), is(localTxId));
assertThat(startedEvent1.type(), is(EventType.TxStartedEvent));
- assertThat(startedEvent1.retries(), is(-1));
+ assertThat(startedEvent1.forwardRetries(), is(-1));
assertThat(startedEvent1.retryMethod(),
is(this.getClass().getDeclaredMethod("doNothing").toString()));
- assertThat(messages.get(1).type(), is(EventType.TxAbortedEvent));
-
- TxEvent startedEvent2 = messages.get(2);
+ TxEvent startedEvent2 = messages.get(1);
assertThat(startedEvent2.localTxId(), is(newLocalTxId));
assertThat(startedEvent2.type(), is(EventType.TxStartedEvent));
- assertThat(startedEvent2.retries(), is(-1));
-
- assertThat(messages.get(3).type(), is(EventType.TxAbortedEvent));
+ assertThat(startedEvent2.forwardRetries(), is(-1));
- TxEvent startedEvent3 = messages.get(4);
+ TxEvent startedEvent3 = messages.get(2);
assertThat(startedEvent3.localTxId(), is(newLocalTxId));
assertThat(startedEvent3.type(), is(EventType.TxStartedEvent));
- assertThat(startedEvent3.retries(), is(-1));
+ assertThat(startedEvent3.forwardRetries(), is(-1));
- assertThat(messages.get(5).type(), is(EventType.TxEndedEvent));
+ assertThat(messages.get(3).type(), is(EventType.TxEndedEvent));
assertThat(omegaContext.globalTxId(), is(globalTxId));
assertThat(omegaContext.localTxId(), is(localTxId));
[servicecomb-pack] 02/03: SCB-1707 Add prefix 'forward' to
parameter retries and timeout
Posted by ni...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ningjiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-pack.git
commit 2259476a3c27c26273d9fada491679bcbeb0628f
Author: Lei Zhang <zh...@apache.org>
AuthorDate: Wed Jan 8 10:13:24 2020 +0800
SCB-1707 Add prefix 'forward' to parameter retries and timeout
---
.../pack/alpha/core/fsm/event/TxStartedEvent.java | 14 ++++++------
.../pack/alpha/fsm/domain/AddTxEventDomain.java | 2 +-
.../pack/alpha/server/GrpcTxEventEndpointImpl.java | 4 ++--
.../alpha/server/fsm/GrpcSagaEventService.java | 4 ++--
.../pack/alpha/server/AlphaIntegrationTest.java | 6 +++---
.../server/AlphaIntegrationWithRandomPortTest.java | 6 +++---
.../alpha/server/fsm/OmegaEventSagaSimulator.java | 6 +++---
.../ExplicitTransactionContextIT.java | 25 ++++++++++------------
.../grpc/saga/GrpcSagaClientMessageSender.java | 8 ++-----
.../grpc/saga/SagaLoadBalancedSenderTestBase.java | 4 ++--
.../spring/TransactionInterceptionTest.java | 18 ++++++----------
.../omega/transaction/AbstractRecoveryPolicy.java | 8 +++----
.../omega/transaction/CompensableInterceptor.java | 4 ++--
.../omega/transaction/EventAwareInterceptor.java | 2 +-
.../pack/omega/transaction/ForwardRecovery.java | 10 ++++-----
.../transaction/NoOpEventAwareInterceptor.java | 2 +-
.../pack/omega/transaction/RecoveryPolicy.java | 2 +-
.../omega/transaction/RecoveryPolicyFactory.java | 4 ++--
.../pack/omega/transaction/TransactionAspect.java | 6 +++---
.../pack/omega/transaction/TxEvent.java | 12 +++++------
.../pack/omega/transaction/TxStartedEvent.java | 4 ++--
.../transaction/CompensableInterceptorTest.java | 6 +++---
.../omega/transaction/DefaultRecoveryTest.java | 6 +++---
.../src/main/proto/GrpcTxEvent.proto | 4 ++--
24 files changed, 77 insertions(+), 90 deletions(-)
diff --git a/alpha/alpha-core/src/main/java/org/apache/servicecomb/pack/alpha/core/fsm/event/TxStartedEvent.java b/alpha/alpha-core/src/main/java/org/apache/servicecomb/pack/alpha/core/fsm/event/TxStartedEvent.java
index 4b6a64b..96630d1 100644
--- a/alpha/alpha-core/src/main/java/org/apache/servicecomb/pack/alpha/core/fsm/event/TxStartedEvent.java
+++ b/alpha/alpha-core/src/main/java/org/apache/servicecomb/pack/alpha/core/fsm/event/TxStartedEvent.java
@@ -24,7 +24,7 @@ public class TxStartedEvent extends TxEvent {
private String compensationMethod;
private byte[] payloads;
private String retryMethod;
- private int retries;
+ private int forwardRetries;
public String getCompensationMethod() {
return compensationMethod;
@@ -50,12 +50,12 @@ public class TxStartedEvent extends TxEvent {
this.retryMethod = retryMethod;
}
- public int getRetries() {
- return retries;
+ public int getForwardRetries() {
+ return forwardRetries;
}
- public void setRetries(int retries) {
- this.retries = retries;
+ public void setForwardRetries(int forwardRetries) {
+ this.forwardRetries = forwardRetries;
}
public static Builder builder() {
@@ -110,8 +110,8 @@ public class TxStartedEvent extends TxEvent {
return this;
}
- public Builder retries(int retries) {
- txStartedEvent.setRetries(retries);
+ public Builder forwardRetries(int forwardRetries) {
+ txStartedEvent.setForwardRetries(forwardRetries);
return this;
}
diff --git a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/domain/AddTxEventDomain.java b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/domain/AddTxEventDomain.java
index 0e410a3..17b0d86 100644
--- a/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/domain/AddTxEventDomain.java
+++ b/alpha/alpha-fsm/src/main/java/org/apache/servicecomb/pack/alpha/fsm/domain/AddTxEventDomain.java
@@ -32,7 +32,7 @@ public class AddTxEventDomain implements DomainEvent {
this.event = event;
this.compensationMethod = event.getCompensationMethod();
this.payloads = event.getPayloads();
- this.retries = event.getRetries();
+ this.retries = event.getForwardRetries();
}
public TxState getState() {
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/GrpcTxEventEndpointImpl.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/GrpcTxEventEndpointImpl.java
index e04a494..5cf1c4a 100644
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/GrpcTxEventEndpointImpl.java
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/GrpcTxEventEndpointImpl.java
@@ -106,9 +106,9 @@ class GrpcTxEventEndpointImpl extends TxEventServiceImplBase {
message.getParentTxId().isEmpty() ? null : message.getParentTxId(),
message.getType(),
message.getCompensationMethod(),
- message.getTimeout(),
+ message.getForwardTimeout(),
message.getRetryMethod(),
- message.getRetries(),
+ message.getForwardRetries(),
message.getPayloads().toByteArray()
));
diff --git a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/fsm/GrpcSagaEventService.java b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/fsm/GrpcSagaEventService.java
index dac6f21..79dac46 100644
--- a/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/fsm/GrpcSagaEventService.java
+++ b/alpha/alpha-server/src/main/java/org/apache/servicecomb/pack/alpha/server/fsm/GrpcSagaEventService.java
@@ -108,7 +108,7 @@ public class GrpcSagaEventService extends TxEventServiceImplBase {
.instanceId(message.getInstanceId())
.globalTxId(message.getGlobalTxId())
.createTime(new Date())
- .timeout(message.getTimeout()).build();
+ .timeout(message.getForwardTimeout()).build();
} else if (message.getType().equals(EventType.SagaEndedEvent.name())) {
event = org.apache.servicecomb.pack.alpha.core.fsm.event.SagaEndedEvent.builder()
.serviceName(message.getServiceName())
@@ -137,7 +137,7 @@ public class GrpcSagaEventService extends TxEventServiceImplBase {
.parentTxId(message.getParentTxId().isEmpty() ? null : message.getParentTxId())
.compensationMethod(message.getCompensationMethod())
.retryMethod(message.getRetryMethod())
- .retries(message.getRetries())
+ .forwardRetries(message.getForwardRetries())
.createTime(new Date())
.payloads(message.getPayloads().toByteArray()).build();
} else if (message.getType().equals(EventType.TxEndedEvent.name())) {
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/AlphaIntegrationTest.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/AlphaIntegrationTest.java
index 09b268b..fabba6c 100644
--- a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/AlphaIntegrationTest.java
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/AlphaIntegrationTest.java
@@ -553,7 +553,7 @@ public class AlphaIntegrationTest {
String compensationMethod,
int timeout,
String retryMethod,
- int retries) {
+ int forwardRetries) {
return GrpcTxEvent.newBuilder()
.setServiceName(serviceName)
@@ -564,9 +564,9 @@ public class AlphaIntegrationTest {
.setParentTxId(parentTxId == null ? "" : parentTxId)
.setType(eventType.name())
.setCompensationMethod(compensationMethod)
- .setTimeout(timeout)
+ .setForwardTimeout(timeout)
.setRetryMethod(retryMethod)
- .setRetries(retries)
+ .setForwardRetries(forwardRetries)
.setPayloads(ByteString.copyFrom(payloads))
.build();
}
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/AlphaIntegrationWithRandomPortTest.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/AlphaIntegrationWithRandomPortTest.java
index 84f12f4..d4c44b6 100644
--- a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/AlphaIntegrationWithRandomPortTest.java
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/AlphaIntegrationWithRandomPortTest.java
@@ -535,7 +535,7 @@ public class AlphaIntegrationWithRandomPortTest {
String compensationMethod,
int timeout,
String retryMethod,
- int retries) {
+ int forwardRetries) {
return GrpcTxEvent.newBuilder()
.setServiceName(serviceName)
@@ -546,9 +546,9 @@ public class AlphaIntegrationWithRandomPortTest {
.setParentTxId(parentTxId == null ? "" : parentTxId)
.setType(eventType.name())
.setCompensationMethod(compensationMethod)
- .setTimeout(timeout)
+ .setForwardTimeout(timeout)
.setRetryMethod(retryMethod)
- .setRetries(retries)
+ .setForwardRetries(forwardRetries)
.setPayloads(ByteString.copyFrom(payloads))
.build();
}
diff --git a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/OmegaEventSagaSimulator.java b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/OmegaEventSagaSimulator.java
index ae6717a..424dd66 100644
--- a/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/OmegaEventSagaSimulator.java
+++ b/alpha/alpha-server/src/test/java/org/apache/servicecomb/pack/alpha/server/fsm/OmegaEventSagaSimulator.java
@@ -255,7 +255,7 @@ public class OmegaEventSagaSimulator {
String compensationMethod,
int timeout,
String retryMethod,
- int retries) {
+ int forwardRetries) {
return GrpcTxEvent.newBuilder()
.setServiceName(serviceName)
@@ -266,9 +266,9 @@ public class OmegaEventSagaSimulator {
.setParentTxId(parentTxId == null ? "" : parentTxId)
.setType(eventType.name())
.setCompensationMethod(compensationMethod)
- .setTimeout(timeout)
+ .setForwardTimeout(timeout)
.setRetryMethod(retryMethod)
- .setRetries(retries)
+ .setForwardRetries(forwardRetries)
.setPayloads(ByteString.copyFrom(payloads))
.build();
}
diff --git a/integration-tests/explicit-transaction-context-tests/src/test/java/org/apache/servicecomb/pack/integration/tests/explicitcontext/ExplicitTransactionContextIT.java b/integration-tests/explicit-transaction-context-tests/src/test/java/org/apache/servicecomb/pack/integration/tests/explicitcontext/ExplicitTransactionContextIT.java
index 4b752b9..dc2bf58 100644
--- a/integration-tests/explicit-transaction-context-tests/src/test/java/org/apache/servicecomb/pack/integration/tests/explicitcontext/ExplicitTransactionContextIT.java
+++ b/integration-tests/explicit-transaction-context-tests/src/test/java/org/apache/servicecomb/pack/integration/tests/explicitcontext/ExplicitTransactionContextIT.java
@@ -255,23 +255,22 @@ public class ExplicitTransactionContextIT {
assertThat(entity.getStatusCode(), is(OK));
assertThat(entity.getBody(), is("Greetings, eric; Welcome to visit the zoo, eric"));
- await().atMost(10, SECONDS).until(() -> eventRepo.count() == 8);
+ await().atMost(10, SECONDS).until(() -> eventRepo.count() == 7);
List<String> distinctGlobalTxIds = eventRepo.findDistinctGlobalTxId();
assertThat(distinctGlobalTxIds.size(), greaterThanOrEqualTo(1));
String globalTxId = distinctGlobalTxIds.get(0);
List<TxEvent> events = eventRepo.findByGlobalTxIdOrderByCreationTime(globalTxId);
- assertThat(events.size(), is(8));
+ assertThat(events.size(), is(7));
assertThat(events.get(0).type(), is("SagaStartedEvent"));
assertThat(events.get(1).type(), is("TxStartedEvent"));
assertThat(events.get(2).type(), is("TxEndedEvent"));
assertThat(events.get(3).type(), is("TxStartedEvent"));
- assertThat(events.get(4).type(), is("TxAbortedEvent"));
- assertThat(events.get(5).type(), is("TxStartedEvent"));
- assertThat(events.get(6).type(), is("TxEndedEvent"));
- assertThat(events.get(7).type(), is("SagaEndedEvent"));
+ assertThat(events.get(4).type(), is("TxStartedEvent"));
+ assertThat(events.get(5).type(), is("TxEndedEvent"));
+ assertThat(events.get(6).type(), is("SagaEndedEvent"));
assertThat(compensatedMessages.isEmpty(), is(true));
}
@@ -286,27 +285,25 @@ public class ExplicitTransactionContextIT {
assertThat(entity.getStatusCode(), is(INTERNAL_SERVER_ERROR));
- await().atMost(10, SECONDS).until(() -> eventRepo.count() == 12);
+ await().atMost(10, SECONDS).until(() -> eventRepo.count() == 10);
List<String> distinctGlobalTxIds = eventRepo.findDistinctGlobalTxId();
assertThat(distinctGlobalTxIds.size(), greaterThanOrEqualTo(1));
String globalTxId = distinctGlobalTxIds.get(0);
List<TxEvent> events = eventRepo.findByGlobalTxIdOrderByCreationTime(globalTxId);
- assertThat(events.size(), is(12));
+ assertThat(events.size(), is(10));
assertThat(events.get(0).type(), is("SagaStartedEvent"));
assertThat(events.get(1).type(), is("TxStartedEvent"));
assertThat(events.get(2).type(), is("TxEndedEvent"));
assertThat(events.get(3).type(), is("TxStartedEvent"));
- assertThat(events.get(4).type(), is("TxAbortedEvent"));
+ assertThat(events.get(4).type(), is("TxStartedEvent"));
assertThat(events.get(5).type(), is("TxStartedEvent"));
assertThat(events.get(6).type(), is("TxAbortedEvent"));
- assertThat(events.get(7).type(), is("TxStartedEvent"));
- assertThat(events.get(8).type(), is("TxAbortedEvent"));
- // This event is for the whole saga event
- assertThat(events.get(9).type(), is("TxAbortedEvent"));
- assertThat(events.get(10).type(), is("TxCompensatedEvent"));
+ assertThat(events.get(7).type(), is("TxAbortedEvent"));
+ assertThat(events.get(8).type(), is("TxCompensatedEvent"));
+ assertThat(events.get(9).type(), is("SagaEndedEvent"));
assertThat(compensatedMessages, Matchers.contains("Goodbye, " + GreetingController.TRESPASSER));
}
diff --git a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/pack/omega/connector/grpc/saga/GrpcSagaClientMessageSender.java b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/pack/omega/connector/grpc/saga/GrpcSagaClientMessageSender.java
index 37996a0..60c9bf2 100644
--- a/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/pack/omega/connector/grpc/saga/GrpcSagaClientMessageSender.java
+++ b/omega/omega-connector/omega-connector-grpc/src/main/java/org/apache/servicecomb/pack/omega/connector/grpc/saga/GrpcSagaClientMessageSender.java
@@ -19,8 +19,6 @@ package org.apache.servicecomb.pack.omega.connector.grpc.saga;
import com.google.protobuf.ByteString;
import io.grpc.ManagedChannel;
-import io.grpc.stub.StreamObserver;
-import java.lang.invoke.MethodHandles;
import org.apache.servicecomb.pack.contract.grpc.ServerMeta;
import org.apache.servicecomb.pack.omega.connector.grpc.core.LoadBalanceContext;
import org.apache.servicecomb.pack.omega.context.ServiceConfig;
@@ -37,8 +35,6 @@ import org.apache.servicecomb.pack.contract.grpc.GrpcTxEvent.Builder;
import org.apache.servicecomb.pack.contract.grpc.TxEventServiceGrpc;
import org.apache.servicecomb.pack.contract.grpc.TxEventServiceGrpc.TxEventServiceBlockingStub;
import org.apache.servicecomb.pack.contract.grpc.TxEventServiceGrpc.TxEventServiceStub;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
public class GrpcSagaClientMessageSender implements SagaMessageSender {
@@ -113,10 +109,10 @@ public class GrpcSagaClientMessageSender implements SagaMessageSender {
.setLocalTxId(event.localTxId())
.setParentTxId(event.parentTxId() == null ? "" : event.parentTxId())
.setType(event.type().name())
- .setTimeout(event.timeout())
+ .setForwardTimeout(event.timeout())
.setCompensationMethod(event.compensationMethod())
.setRetryMethod(event.retryMethod() == null ? "" : event.retryMethod())
- .setRetries(event.retries())
+ .setForwardRetries(event.forwardRetries())
.setPayloads(payloads);
return builder.build();
diff --git a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/pack/omega/connector/grpc/saga/SagaLoadBalancedSenderTestBase.java b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/pack/omega/connector/grpc/saga/SagaLoadBalancedSenderTestBase.java
index 2f19080..f2f05d6 100644
--- a/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/pack/omega/connector/grpc/saga/SagaLoadBalancedSenderTestBase.java
+++ b/omega/omega-connector/omega-connector-grpc/src/test/java/org/apache/servicecomb/pack/omega/connector/grpc/saga/SagaLoadBalancedSenderTestBase.java
@@ -171,9 +171,9 @@ public abstract class SagaLoadBalancedSenderTestBase {
request.getLocalTxId(),
request.getParentTxId(),
request.getCompensationMethod(),
- request.getTimeout(),
+ request.getForwardTimeout(),
request.getRetryMethod(),
- request.getRetries(),
+ request.getForwardRetries(),
new String(request.getPayloads().toByteArray())));
sleep();
diff --git a/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/pack/omega/transaction/spring/TransactionInterceptionTest.java b/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/pack/omega/transaction/spring/TransactionInterceptionTest.java
index a5f3abf..81d79a1 100644
--- a/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/pack/omega/transaction/spring/TransactionInterceptionTest.java
+++ b/omega/omega-spring-tx/src/test/java/org/apache/servicecomb/pack/omega/transaction/spring/TransactionInterceptionTest.java
@@ -203,19 +203,16 @@ public class TransactionInterceptionTest {
fail("unexpected exception throw: " + e);
}
- assertThat(messages.size(), is(4));
+ assertThat(messages.size(), is(3));
assertThat(messages.get(0),
is(new TxStartedEvent(globalTxId, newLocalTxId, globalTxId, compensationMethod2, 0, retryMethod, 2, user, 1)
.toString()));
- String abortedEvent = messages.get(1);
- assertThat(abortedEvent, allOf(containsString("TxAbortedEvent"), containsString("Retry harder")));
-
- assertThat(messages.get(2),
+ assertThat(messages.get(1),
is(new TxStartedEvent(globalTxId, newLocalTxId, globalTxId, compensationMethod2, 0, retryMethod, 1, user, 1)
.toString()));
- assertThat(messages.get(3),
+ assertThat(messages.get(2),
is(new TxEndedEvent(globalTxId, newLocalTxId, globalTxId, compensationMethod2).toString()));
assertThat(userRepository.count(), is(1L));
@@ -234,19 +231,16 @@ public class TransactionInterceptionTest {
assertThat(e.getMessage(), is("Retry harder"));
}
- assertThat(messages.size(), is(4));
+ assertThat(messages.size(), is(3));
assertThat(messages.get(0),
is(new TxStartedEvent(globalTxId, newLocalTxId, globalTxId, compensationMethod2, 0, retryMethod, 2, user, 3)
.toString()));
- String abortedEvent1 = messages.get(1);
- assertThat(abortedEvent1, allOf(containsString("TxAbortedEvent"), containsString("Retry harder")));
-
- assertThat(messages.get(2),
+ assertThat(messages.get(1),
is(new TxStartedEvent(globalTxId, newLocalTxId, globalTxId, compensationMethod2, 0, retryMethod, 1, user, 3)
.toString()));
- String abortedEvent2 = messages.get(3);
+ String abortedEvent2 = messages.get(2);
assertThat(abortedEvent2, allOf(containsString("TxAbortedEvent"), containsString("Retry harder")));
assertThat(userRepository.count(), is(0L));
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/AbstractRecoveryPolicy.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/AbstractRecoveryPolicy.java
index d38dec8..00f26ca 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/AbstractRecoveryPolicy.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/AbstractRecoveryPolicy.java
@@ -25,19 +25,19 @@ import org.aspectj.lang.ProceedingJoinPoint;
public abstract class AbstractRecoveryPolicy implements RecoveryPolicy {
public abstract Object applyTo(ProceedingJoinPoint joinPoint, Compensable compensable,
- CompensableInterceptor interceptor, OmegaContext context, String parentTxId, int retries)
+ CompensableInterceptor interceptor, OmegaContext context, String parentTxId, int forwardRetries)
throws Throwable;
@Override
public Object apply(ProceedingJoinPoint joinPoint, Compensable compensable,
- CompensableInterceptor interceptor, OmegaContext context, String parentTxId, int retries)
+ CompensableInterceptor interceptor, OmegaContext context, String parentTxId, int forwardRetries)
throws Throwable {
Object result;
if(compensable.forwardTimeout()>0){
RecoveryPolicyTimeoutWrapper wrapper = new RecoveryPolicyTimeoutWrapper(this);
- result = wrapper.applyTo(joinPoint, compensable, interceptor, context, parentTxId, retries);
+ result = wrapper.applyTo(joinPoint, compensable, interceptor, context, parentTxId, forwardRetries);
} else {
- result = this.applyTo(joinPoint, compensable, interceptor, context, parentTxId, retries);
+ result = this.applyTo(joinPoint, compensable, interceptor, context, parentTxId, forwardRetries);
}
return result;
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/CompensableInterceptor.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/CompensableInterceptor.java
index 108ebbc..d90d63c 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/CompensableInterceptor.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/CompensableInterceptor.java
@@ -30,9 +30,9 @@ public class CompensableInterceptor implements EventAwareInterceptor {
@Override
public AlphaResponse preIntercept(String parentTxId, String compensationMethod, int timeout, String retriesMethod,
- int retries, Object... message) {
+ int forwardRetries, Object... message) {
return sender.send(new TxStartedEvent(context.globalTxId(), context.localTxId(), parentTxId, compensationMethod,
- timeout, retriesMethod, retries, message));
+ timeout, retriesMethod, forwardRetries, message));
}
@Override
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/EventAwareInterceptor.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/EventAwareInterceptor.java
index 5a88ee3..3b4036c 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/EventAwareInterceptor.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/EventAwareInterceptor.java
@@ -21,7 +21,7 @@ public interface EventAwareInterceptor {
AlphaResponse preIntercept(String parentTxId, String compensationMethod, int timeout,
String retriesMethod,
- int retries, Object... message);
+ int forwardRetries, Object... message);
void postIntercept(String parentTxId, String compensationMethod);
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/ForwardRecovery.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/ForwardRecovery.java
index a514eef..21fde8e 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/ForwardRecovery.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/ForwardRecovery.java
@@ -40,9 +40,9 @@ public class ForwardRecovery extends DefaultRecovery {
// TODO: 2018/03/10 we do not support retry with forward timeout yet
@Override
public Object applyTo(ProceedingJoinPoint joinPoint, Compensable compensable, CompensableInterceptor interceptor,
- OmegaContext context, String parentTxId, int retries) throws Throwable {
+ OmegaContext context, String parentTxId, int forwardRetries) throws Throwable {
Method method = ((MethodSignature) joinPoint.getSignature()).getMethod();
- int remains = retries;
+ int remains = forwardRetries;
try {
while (true) {
try {
@@ -55,12 +55,12 @@ public class ForwardRecovery extends DefaultRecovery {
remains = remains == -1 ? -1 : remains - 1;
if (remains == 0) {
LOG.error(
- "Retried sub tx failed maximum times, global tx id: {}, local tx id: {}, method: {}, retried times: {}",
- context.globalTxId(), context.localTxId(), method.toString(), retries);
+ "Forward Retried sub tx failed maximum times, global tx id: {}, local tx id: {}, method: {}, retried times: {}",
+ context.globalTxId(), context.localTxId(), method.toString(), forwardRetries);
throw throwable;
}
- LOG.warn("Retrying sub tx failed, global tx id: {}, local tx id: {}, method: {}, remains: {}",
+ LOG.warn("Forward Retrying sub tx failed, global tx id: {}, local tx id: {}, method: {}, remains: {}",
context.globalTxId(), context.localTxId(), method.toString(), remains);
Thread.sleep(compensable.retryDelayInMilliseconds());
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/NoOpEventAwareInterceptor.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/NoOpEventAwareInterceptor.java
index bc46242..a6f8447 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/NoOpEventAwareInterceptor.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/NoOpEventAwareInterceptor.java
@@ -24,7 +24,7 @@ public class NoOpEventAwareInterceptor implements EventAwareInterceptor {
@Override
public AlphaResponse preIntercept(String parentTxId, String compensationMethod, int timeout,
String retriesMethod,
- int retries, Object... message) {
+ int forwardRetries, Object... message) {
return new AlphaResponse(false);
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/RecoveryPolicy.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/RecoveryPolicy.java
index 7aec7c7..b2ab04f 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/RecoveryPolicy.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/RecoveryPolicy.java
@@ -23,5 +23,5 @@ import org.aspectj.lang.ProceedingJoinPoint;
public interface RecoveryPolicy {
Object apply(ProceedingJoinPoint joinPoint, Compensable compensable, CompensableInterceptor interceptor,
- OmegaContext context, String parentTxId, int retries) throws Throwable;
+ OmegaContext context, String parentTxId, int forwardRetries) throws Throwable;
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/RecoveryPolicyFactory.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/RecoveryPolicyFactory.java
index 33098d9..f373067 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/RecoveryPolicyFactory.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/RecoveryPolicyFactory.java
@@ -27,7 +27,7 @@ public class RecoveryPolicyFactory {
* If retries > 0, it will use the forward recovery and retry the given times at most.
* If retries == -1, it will use the forward recovery and retry forever until interrupted.
*/
- static RecoveryPolicy getRecoveryPolicy(int retries) {
- return retries != 0 ? FORWARD_RECOVERY : DEFAULT_RECOVERY;
+ static RecoveryPolicy getRecoveryPolicy(int forwardRetries) {
+ return forwardRetries != 0 ? FORWARD_RECOVERY : DEFAULT_RECOVERY;
}
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/TransactionAspect.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/TransactionAspect.java
index 826480d..60efca0 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/TransactionAspect.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/TransactionAspect.java
@@ -63,10 +63,10 @@ public class TransactionAspect extends TransactionContextHelper {
context.newLocalTxId();
LOG.debug("Updated context {} for compensable method {} ", context, method.toString());
- int retries = compensable.forwardRetries();
- RecoveryPolicy recoveryPolicy = RecoveryPolicyFactory.getRecoveryPolicy(retries);
+ int forwardRetries = compensable.forwardRetries();
+ RecoveryPolicy recoveryPolicy = RecoveryPolicyFactory.getRecoveryPolicy(forwardRetries);
try {
- return recoveryPolicy.apply(joinPoint, compensable, interceptor, context, localTxId, retries);
+ return recoveryPolicy.apply(joinPoint, compensable, interceptor, context, localTxId, forwardRetries);
} finally {
context.setLocalTxId(localTxId);
LOG.debug("Restored context back to {}", context);
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/TxEvent.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/TxEvent.java
index 71e10b5..0bc5c26 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/TxEvent.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/TxEvent.java
@@ -33,10 +33,10 @@ public class TxEvent {
private final Object[] payloads;
private final String retryMethod;
- private final int retries;
+ private final int forwardRetries;
public TxEvent(EventType type, String globalTxId, String localTxId, String parentTxId, String compensationMethod,
- int timeout, String retryMethod, int retries, Object... payloads) {
+ int timeout, String retryMethod, int forwardRetries, Object... payloads) {
this.timestamp = System.currentTimeMillis();
this.type = type;
this.globalTxId = globalTxId;
@@ -45,7 +45,7 @@ public class TxEvent {
this.compensationMethod = compensationMethod;
this.timeout = timeout;
this.retryMethod = retryMethod;
- this.retries = retries;
+ this.forwardRetries = forwardRetries;
this.payloads = payloads;
}
@@ -85,8 +85,8 @@ public class TxEvent {
return retryMethod;
}
- public int retries() {
- return retries;
+ public int forwardRetries() {
+ return forwardRetries;
}
@Override
@@ -98,7 +98,7 @@ public class TxEvent {
", compensationMethod='" + compensationMethod + '\'' +
", timeout=" + timeout +
", retryMethod='" + retryMethod + '\'' +
- ", retries=" + retries +
+ ", forwardRetries=" + forwardRetries +
", payloads=" + Arrays.toString(payloads) +
'}';
}
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/TxStartedEvent.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/TxStartedEvent.java
index fe7bb5b..0851f65 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/TxStartedEvent.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/TxStartedEvent.java
@@ -22,8 +22,8 @@ import org.apache.servicecomb.pack.common.EventType;
public class TxStartedEvent extends TxEvent {
public TxStartedEvent(String globalTxId, String localTxId, String parentTxId, String compensationMethod,
- int timeout, String retryMethod, int retries, Object... payloads) {
+ int timeout, String retryMethod, int forwardRetries, Object... payloads) {
super(EventType.TxStartedEvent, globalTxId, localTxId, parentTxId, compensationMethod, timeout, retryMethod,
- retries, payloads);
+ forwardRetries, payloads);
}
}
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/CompensableInterceptorTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/CompensableInterceptorTest.java
index 02c2067..fa3385c 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/CompensableInterceptorTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/CompensableInterceptorTest.java
@@ -91,15 +91,15 @@ public class CompensableInterceptorTest {
@Test
public void sendsTxStartedEventBefore() throws Exception {
- int retries = new Random().nextInt();
- interceptor.preIntercept(parentTxId, compensationMethod, 0, retryMethod, retries, message);
+ int forwardRetries = new Random().nextInt();
+ interceptor.preIntercept(parentTxId, compensationMethod, 0, retryMethod, forwardRetries, message);
TxEvent event = messages.get(0);
assertThat(event.globalTxId(), is(globalTxId));
assertThat(event.localTxId(), is(localTxId));
assertThat(event.parentTxId(), is(parentTxId));
- assertThat(event.retries(), is(retries));
+ assertThat(event.forwardRetries(), is(forwardRetries));
assertThat(event.retryMethod(), is(retryMethod));
assertThat(event.type(), is(EventType.TxStartedEvent));
assertThat(event.compensationMethod(), is(compensationMethod));
diff --git a/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/DefaultRecoveryTest.java b/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/DefaultRecoveryTest.java
index 23b29fa..0bcf0de 100644
--- a/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/DefaultRecoveryTest.java
+++ b/omega/omega-transaction/src/test/java/org/apache/servicecomb/pack/omega/transaction/DefaultRecoveryTest.java
@@ -131,7 +131,7 @@ public class DefaultRecoveryTest {
assertThat(startedEvent.localTxId(), is(localTxId));
assertThat(startedEvent.parentTxId(), is(parentTxId));
assertThat(startedEvent.type(), is(EventType.TxStartedEvent));
- assertThat(startedEvent.retries(), is(0));
+ assertThat(startedEvent.forwardRetries(), is(0));
assertThat(startedEvent.retryMethod(), is(""));
TxEvent endedEvent = messages.get(1);
@@ -159,7 +159,7 @@ public class DefaultRecoveryTest {
assertThat(startedEvent.localTxId(), is(localTxId));
assertThat(startedEvent.parentTxId(), is(parentTxId));
assertThat(startedEvent.type(), is(EventType.TxStartedEvent));
- assertThat(startedEvent.retries(), is(0));
+ assertThat(startedEvent.forwardRetries(), is(0));
assertThat(startedEvent.retryMethod(), is(""));
TxEvent abortedEvent = messages.get(1);
@@ -201,7 +201,7 @@ public class DefaultRecoveryTest {
assertThat(startedEvent.localTxId(), is(localTxId));
assertThat(startedEvent.parentTxId(), is(parentTxId));
assertThat(startedEvent.type(), is(EventType.TxStartedEvent));
- assertThat(startedEvent.retries(), is(retries));
+ assertThat(startedEvent.forwardRetries(), is(retries));
assertThat(startedEvent.retryMethod(), is(this.getClass().getDeclaredMethod("doNothing").toString()));
}
diff --git a/pack-contracts/pack-contract-grpc/src/main/proto/GrpcTxEvent.proto b/pack-contracts/pack-contract-grpc/src/main/proto/GrpcTxEvent.proto
index dd3a73c..6d34d21 100644
--- a/pack-contracts/pack-contract-grpc/src/main/proto/GrpcTxEvent.proto
+++ b/pack-contracts/pack-contract-grpc/src/main/proto/GrpcTxEvent.proto
@@ -42,8 +42,8 @@ message GrpcTxEvent {
bytes payloads = 7;
string serviceName = 8;
string instanceId = 9;
- int32 timeout = 10;
- int32 retries = 11;
+ int32 forwardTimeout = 10;
+ int32 forwardRetries = 11;
string retryMethod = 12;
}
[servicecomb-pack] 01/03: SCB-1707 forward compensation only sends
once a failure event if multiple retries fail
Posted by ni...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ningjiang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/servicecomb-pack.git
commit cac096b3a4e24aa5b5c8b631c0b5ff3dcae4d73e
Author: Lei Zhang <zh...@apache.org>
AuthorDate: Wed Jan 8 10:06:47 2020 +0800
SCB-1707 forward compensation only sends once a failure event if multiple retries fail
---
.../servicecomb/pack/omega/transaction/DefaultRecovery.java | 11 +++++++----
1 file changed, 7 insertions(+), 4 deletions(-)
diff --git a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/DefaultRecovery.java b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/DefaultRecovery.java
index 3418f3a..19e7825 100644
--- a/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/DefaultRecovery.java
+++ b/omega/omega-transaction/src/main/java/org/apache/servicecomb/pack/omega/transaction/DefaultRecovery.java
@@ -47,17 +47,17 @@ public class DefaultRecovery extends AbstractRecoveryPolicy {
@Override
public Object applyTo(ProceedingJoinPoint joinPoint, Compensable compensable, CompensableInterceptor interceptor,
- OmegaContext context, String parentTxId, int retries) throws Throwable {
+ OmegaContext context, String parentTxId, int forwardRetries) throws Throwable {
Method method = ((MethodSignature) joinPoint.getSignature()).getMethod();
LOG.debug("Intercepting compensable method {} with context {}", method.toString(), context);
String compensationSignature =
compensable.compensationMethod().isEmpty() ? "" : compensationMethodSignature(joinPoint, compensable, method);
- String retrySignature = (retries != 0 || compensationSignature.isEmpty()) ? method.toString() : "";
+ String retrySignature = (forwardRetries != 0 || compensationSignature.isEmpty()) ? method.toString() : "";
AlphaResponse response = interceptor.preIntercept(parentTxId, compensationSignature, compensable.forwardTimeout(),
- retrySignature, retries, joinPoint.getArgs());
+ retrySignature, forwardRetries, joinPoint.getArgs());
if (response.aborted()) {
String abortedLocalTxId = context.localTxId();
context.setLocalTxId(parentTxId);
@@ -71,7 +71,10 @@ public class DefaultRecovery extends AbstractRecoveryPolicy {
return result;
} catch (Throwable throwable) {
- interceptor.onError(parentTxId, compensationSignature, throwable);
+ if (compensable.forwardRetries() == 0 || (compensable.forwardRetries() > 0
+ && forwardRetries == 1)) {
+ interceptor.onError(parentTxId, compensationSignature, throwable);
+ }
throw throwable;
}
}