This is an automated email from the ASF dual-hosted git repository.
wangjian pushed a commit to branch 2.x
in repository https://gitbox.apache.org/repos/asf/incubator-seata.git
The following commit(s) were added to refs/heads/2.x by this push:
new 1adb847e6f feature: add detailed saga result reporting to benchmark
cli (#8042)
1adb847e6f is described below
commit 1adb847e6f722a1fc0e0259f7f68fbb5c5aa1ea6
Author: zihenzzz <[email protected]>
AuthorDate: Wed Apr 15 18:38:17 2026 +0800
feature: add detailed saga result reporting to benchmark cli (#8042)
---
changes/en-us/2.x.md | 1 +
changes/zh-cn/2.x.md | 1 +
test-suite/seata-benchmark-cli/README.md | 190 ++++++++++++++-
.../seata/benchmark/BenchmarkApplication.java | 61 ++++-
.../apache/seata/benchmark/BenchmarkRunner.java | 2 +-
.../seata/benchmark/config/BenchmarkConfig.java | 110 +++++++++
.../benchmark/config/BenchmarkConfigLoader.java | 26 ++-
.../benchmark/constant/BenchmarkConstants.java | 3 +
.../seata/benchmark/executor/SagaModeExecutor.java | 254 +++++++++++++++++++--
.../benchmark/executor/WorkloadGenerator.java | 13 +-
.../seata/benchmark/model/BenchmarkMetrics.java | 81 ++++++-
.../seata/benchmark/monitor/MetricsCollector.java | 40 +++-
.../benchmark/saga/FailureRandomProvider.java | 51 +++++
.../benchmark/saga/InventoryDbSagaService.java | 183 +++++++++++++++
.../seata/benchmark/saga/InventorySagaService.java | 50 +++-
.../seata/benchmark/saga/OrderDbSagaService.java | 158 +++++++++++++
.../seata/benchmark/saga/OrderSagaService.java | 50 +++-
.../seata/benchmark/saga/PaymentDbSagaService.java | 179 +++++++++++++++
.../seata/benchmark/saga/PaymentSagaService.java | 50 +++-
.../seata/benchmark/saga/SagaDbEnvironment.java | 171 ++++++++++++++
20 files changed, 1621 insertions(+), 53 deletions(-)
diff --git a/changes/en-us/2.x.md b/changes/en-us/2.x.md
index 6b09014774..3fcbb47b13 100644
--- a/changes/en-us/2.x.md
+++ b/changes/en-us/2.x.md
@@ -43,6 +43,7 @@ Add changes here for all PR submitted to the 2.x branch.
### optimize:
+- [[#8042](https://github.com/apache/incubator-seata/pull/8042)] enhance the
traditional Saga benchmark with a more realistic state-machine workload,
reproducible failure modeling, richer Saga result reporting, and clearer Saga
parameter semantics
- [[#8022](https://github.com/apache/incubator-seata/pull/8022)] configure PMD
to output messages in English
- [[#7930](https://github.com/apache/incubator-seata/pull/7930)] pin the
Spring version for namingserver and console
- [[#7943](https://github.com/apache/incubator-seata/pull/7943)] update
jib-maven-plugin version and increase parallel test execution limits
diff --git a/changes/zh-cn/2.x.md b/changes/zh-cn/2.x.md
index 4f3a8bf2d8..2f83fff256 100644
--- a/changes/zh-cn/2.x.md
+++ b/changes/zh-cn/2.x.md
@@ -45,6 +45,7 @@
### optimize:
+- [[#8042](https://github.com/apache/incubator-seata/pull/8042)] 增强传统 Saga
benchmark，支持更真实的状态机工作负载、可复现的失败建模、更丰富的 Saga 结果报告，以及更清晰的 Saga 参数语义
- [[#8022](https://github.com/apache/incubator-seata/pull/8022)] 配置PMD输出英文消息
- [[#7930](https://github.com/apache/incubator-seata/pull/7930)]
固定namingserver和console的Spring版本
- [[#7943](https://github.com/apache/incubator-seata/pull/7943)]
升级jib-maven-plugin版本和提高ci并行度
diff --git a/test-suite/seata-benchmark-cli/README.md
b/test-suite/seata-benchmark-cli/README.md
index 5a47136027..49ac85fe9c 100644
--- a/test-suite/seata-benchmark-cli/README.md
+++ b/test-suite/seata-benchmark-cli/README.md
@@ -27,6 +27,11 @@ A command-line benchmark tool for stress testing Seata
transaction modes.
- **Configurable TPS** (Transactions Per Second) control
- **Multi-threaded** workload generation
- **Fault injection** with configurable rollback percentage
+- **Selectable SAGA state machine shapes** such as `simple` and `order`
+- **Selectable SAGA workload implementations** such as `mock` and `db`
+- **Step-targeted SAGA failure injection** for forward steps such as
`inventory`, `payment`, and `order`
+- **Reproducible SAGA failure injection** with `--saga-random-seed`
+- **Step-targeted SAGA timeout simulation** with `--saga-timeout-step` and
`--saga-timeout-ms`
- **Window-based progress reporting** (every 10 seconds)
- Performance metrics collection (latency percentiles, success rate, TPS)
- **CSV export** for post-analysis
@@ -96,6 +101,59 @@ java -jar seata-benchmark-cli.jar \
--duration 60 \
--branches 3 \
--rollback-percentage 5
+
+# SAGA mode with explicit order state machine shape
+java -jar seata-benchmark-cli.jar \
+ --server 127.0.0.1:8091 \
+ --mode SAGA \
+ --tps 100 \
+ --duration 60 \
+ --branches 3 \
+ --saga-shape order
+
+# SAGA mode with DB-backed business actions (via Testcontainers MySQL)
+java -jar seata-benchmark-cli.jar \
+ --server 127.0.0.1:8091 \
+ --mode SAGA \
+ --tps 100 \
+ --duration 60 \
+ --branches 3 \
+ --saga-shape order \
+ --saga-workload db
+
+# SAGA mode with payment-step failure injection
+java -jar seata-benchmark-cli.jar \
+ --server 127.0.0.1:8091 \
+ --mode SAGA \
+ --tps 100 \
+ --duration 60 \
+ --branches 3 \
+ --rollback-percentage 20 \
+ --saga-fail-step payment
+
+# SAGA mode with reproducible payment-step failure injection
+java -jar seata-benchmark-cli.jar \
+ --server 127.0.0.1:8091 \
+ --mode SAGA \
+ --tps 100 \
+ --duration 60 \
+ --branches 3 \
+ --saga-shape order \
+ --rollback-percentage 20 \
+ --saga-fail-step payment \
+ --saga-random-seed 123
+
+# SAGA mode with payment-step timeout simulation
+java -jar seata-benchmark-cli.jar \
+ --server 127.0.0.1:8091 \
+ --mode SAGA \
+ --tps 10 \
+ --threads 1 \
+ --duration 10 \
+ --branches 3 \
+ --saga-shape order \
+ --saga-timeout-step payment \
+ --saga-timeout-ms 3000
```
### Performance Testing Modes
@@ -157,6 +215,12 @@ java -jar seata-benchmark-cli.jar \
Usage: seata-benchmark [-hV] [--application-id=<applicationId>]
[-d=<duration>] [--export-csv=<exportCsv>]
[-m=<mode>] [-s=<server>] [-t=<targetTps>]
+ [--saga-shape=<sagaShape>]
+ [--saga-workload=<sagaWorkload>]
+ [--saga-fail-step=<sagaFailStep>]
+ [--saga-random-seed=<sagaRandomSeed>]
+ [--saga-timeout-step=<sagaTimeoutStep>]
+ [--saga-timeout-ms=<sagaTimeoutMs>]
[--threads=<threads>]
[--tx-service-group=<txServiceGroup>]
[--warmup-duration=<warmupDuration>]
[--rollback-percentage=<rollbackPercentage>]
@@ -172,6 +236,26 @@ Options:
Warmup duration in seconds (default: 0)
--rollback-percentage=<rollbackPercentage>
Rollback percentage for fault injection
(0-100, default: 2)
+ --saga-shape=<sagaShape> Select SAGA state machine shape: simple
or order.
+ If omitted, the benchmark keeps the
existing
+ branches-based compatibility behavior.
+ --saga-workload=<sagaWorkload> Select SAGA workload implementation:
mock or db.
+ The default is mock. The db workload
uses
+ Testcontainers MySQL for DB-backed
order,
+ inventory, and payment actions.
+ --saga-fail-step=<sagaFailStep> Restrict SAGA failure injection to one
forward step:
+ inventory, payment, or order.
+ The failure ratio is still controlled by
+ --rollback-percentage.
+ --saga-random-seed=<sagaRandomSeed>
+ Optional random seed for reproducible
SAGA
+ failure injection behavior.
+ --saga-timeout-step=<sagaTimeoutStep>
+ Simulate SAGA timeout at one forward
step:
+ inventory, payment, or order.
+ --saga-timeout-ms=<sagaTimeoutMs>
+ Simulated timeout delay in milliseconds
for
+ SAGA timeout injection (default: 3000).
--branches=<branches> Number of branch transactions
0 = empty mode (protocol overhead only)
>=1 = real mode (actual execution)
@@ -232,21 +316,37 @@ When the benchmark completes, a final report is displayed:
===================================================
Seata Benchmark Final Report
===================================================
+Mode: SAGA
+Saga Workload: db
+Saga Shape: order
Total Transactions: 6,000
-Success Count: 5,940
-Failed Count: 60
-Success Rate: 99.00%
-Average TPS: 100.2
+Success Count: 5,780
+Failed Count: 220
+Success Rate: 96.33%
+Committed Count: 4,860
+Compensated Count: 920
+Execution Failed Count: 180
+Compensation Failed Count: 40
+Unknown Count: 0
+Committed Rate: 81.00%
+Compensated Rate: 15.33%
+End-State Success Rate: 96.33%
+Average TPS: 100.0
Elapsed Time: 60 seconds
Latency Statistics:
P50: 12 ms
P95: 45 ms
P99: 89 ms
+ P99.9: 120 ms
Max: 230 ms
===================================================
```
+For timeout simulation scenarios, the reported `Elapsed Time` may slightly
exceed the configured `--duration` because already-started transactions are
allowed to finish before the workload generator stops.
+
+For DB-backed SAGA scenarios, the final report and CSV output also include
`Saga Workload`, so benchmark results from `mock` and `db` workloads can be
compared explicitly.
+
### CSV Export
Use `--export-csv` to export metrics:
@@ -264,17 +364,29 @@ Output format:
```csv
Metric,Value
+Mode,SAGA
+Saga Workload,db
+Saga Shape,order
Total Transactions,6000
-Success Count,5940
-Failed Count,60
-Success Rate (%),99.00
-Average TPS,100.2
+Success Count,5780
+Failed Count,220
+Success Rate (%),96.33
+Committed Count,4860
+Compensated Count,920
+Execution Failed Count,180
+Compensation Failed Count,40
+Unknown Count,0
+Committed Rate (%),81.00
+Compensated Rate (%),15.33
+End-State Success Rate (%),96.33
+Average TPS,100.0
Elapsed Time (s),60
Latency P50 (ms),12
Latency P95 (ms),45
Latency P99 (ms),89
+Latency P99.9 (ms),120
Latency Max (ms),230
-Export Timestamp,2025-12-01 10:30:45
+Export Time,2025-12-01 10:30:45
```
## Examples
@@ -313,6 +425,48 @@ java -jar seata-benchmark-cli.jar \
--rollback-percentage 5
```
+### Test SAGA Mode with Payment-Step Failure Injection
+
+```bash
+java -jar seata-benchmark-cli.jar \
+ --server 127.0.0.1:8091 \
+ --mode SAGA \
+ --tps 100 \
+ --duration 60 \
+ --branches 3 \
+ --rollback-percentage 20 \
+ --saga-fail-step payment
+```
+
+### Test SAGA Mode with DB-Backed Business Actions
+
+```bash
+java -jar seata-benchmark-cli.jar \
+ --server 127.0.0.1:8091 \
+ --mode SAGA \
+ --tps 100 \
+ --duration 60 \
+ --branches 3 \
+ --saga-shape order \
+ --saga-workload db
+```
+
+### Test DB-Backed SAGA with Payment-Step Failure Injection
+
+```bash
+java -jar seata-benchmark-cli.jar \
+ --server 127.0.0.1:8091 \
+ --mode SAGA \
+ --tps 10 \
+ --threads 1 \
+ --duration 10 \
+ --branches 3 \
+ --saga-shape order \
+ --saga-workload db \
+ --rollback-percentage 20 \
+ --saga-fail-step payment
+```
+
### Test TCC Mode at High Load
```bash
@@ -372,10 +526,28 @@ When `--branches` is set to a value greater than 0:
- **SAGA Mode**:
- Uses Seata state machine engine
- Executes predefined state machine definitions
+ - Supports `mock` and `db` workloads
- Supports compensation on failure
- Available state machines:
- `benchmarkSimpleSaga`: For 1-2 branches
- `benchmarkOrderSaga`: For 3+ branches (order/inventory/payment)
+ - DB workload behavior:
+ - Starts MySQL via Testcontainers
+ - Creates benchmark inventory, account, and order tables
+ - Executes DB-backed inventory/payment/order actions with compensation
+
+### SAGA Workloads
+
+- `mock` workload:
+ - Keeps the lightweight benchmark-oriented implementation
+ - Uses in-memory Saga services with simulated delay, failure injection, and
timeout injection
+ - Is the default workload and preserves backward compatibility
+
+- `db` workload:
+ - Starts a MySQL container via Testcontainers
+ - Initializes benchmark tables for inventory, account, and order data
+ - Executes DB-backed order, inventory, and payment actions while still
supporting the same Saga shape, fail-step, random-seed, and timeout options
+ - Is intended for more realistic business-style Saga benchmarking
### Fault Injection
diff --git
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/BenchmarkApplication.java
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/BenchmarkApplication.java
index 3018d7ccbe..3d6987c2c4 100644
---
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/BenchmarkApplication.java
+++
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/BenchmarkApplication.java
@@ -19,6 +19,7 @@ package org.apache.seata.benchmark;
import org.apache.seata.benchmark.config.BenchmarkConfig;
import org.apache.seata.benchmark.config.BenchmarkConfigLoader;
import org.apache.seata.benchmark.constant.BenchmarkConstants;
+import org.apache.seata.core.model.BranchType;
import picocli.CommandLine;
import picocli.CommandLine.Command;
import picocli.CommandLine.Option;
@@ -93,6 +94,36 @@ public class BenchmarkApplication implements
Callable<Integer> {
description = "Number of branch transactions (0=empty mode,
>=1=real mode with actual execution)")
private Integer branches;
+ @Option(
+ names = {"--saga-shape"},
+ description = "Select SAGA state machine shape: simple or order")
+ private String sagaShape;
+
+ @Option(
+ names = {"--saga-workload"},
+ description = "Select SAGA workload implementation: mock or db")
+ private String sagaWorkload;
+
+ @Option(
+ names = {"--saga-fail-step"},
+ description = "Force SAGA forward failure at a specific step:
inventory, payment, or order")
+ private String sagaFailStep;
+
+ @Option(
+ names = {"--saga-random-seed"},
+ description = "Seed for reproducible SAGA failure injection")
+ private Long sagaRandomSeed;
+
+ @Option(
+ names = {"--saga-timeout-step"},
+ description = "Simulate SAGA timeout at a specific forward step:
inventory, payment, or order")
+ private String sagaTimeoutStep;
+
+ @Option(
+ names = {"--saga-timeout-ms"},
+ description = "Simulated timeout delay in milliseconds for SAGA
timeout injection (default: 3000)")
+ private Integer sagaTimeoutMs;
+
public static void main(String[] args) {
// Parse server address from args before any Seata class loading
String serverAddr = BenchmarkConstants.DEFAULT_SERVER_ADDRESS;
@@ -143,7 +174,13 @@ public class BenchmarkApplication implements
Callable<Integer> {
applicationId,
txServiceGroup,
rollbackPercentage,
- branches);
+ branches,
+ sagaShape,
+ sagaWorkload,
+ sagaFailStep,
+ sagaRandomSeed,
+ sagaTimeoutStep,
+ sagaTimeoutMs);
}
private void printConfiguration(BenchmarkConfig config) {
@@ -159,6 +196,28 @@ public class BenchmarkApplication implements
Callable<Integer> {
System.out.println(" Rollback %: " + config.getRollbackPercentage()
+ "%");
System.out.println(" Branches: " + config.getBranches()
+ (config.getBranches() == 0 ? " (empty mode)" : " (real
mode)"));
+ if (config.getMode() == BranchType.SAGA
+ && config.getSagaWorkload() != null
+ && !config.getSagaWorkload().isEmpty()) {
+ System.out.println(" Saga Workload:" +
padValue(config.getSagaWorkload()));
+ }
+ if (config.getSagaShape() != null && !config.getSagaShape().isEmpty())
{
+ System.out.println(" Saga Shape: " + config.getSagaShape());
+ }
+ if (config.getSagaFailStep() != null &&
!config.getSagaFailStep().isEmpty()) {
+ System.out.println(" Saga Fail: " + config.getSagaFailStep());
+ }
+ if (config.getSagaRandomSeed() != null) {
+ System.out.println(" Saga Seed: " +
config.getSagaRandomSeed());
+ }
+ if (config.getSagaTimeoutStep() != null &&
!config.getSagaTimeoutStep().isEmpty()) {
+ System.out.println(" Saga Timeout: " +
config.getSagaTimeoutStep());
+ System.out.println(" Timeout Ms: " + config.getSagaTimeoutMs());
+ }
System.out.println();
}
+
+ private String padValue(String value) {
+ return value == null ? "" : " " + value;
+ }
}
diff --git
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/BenchmarkRunner.java
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/BenchmarkRunner.java
index 929b59ec88..da013e734f 100644
---
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/BenchmarkRunner.java
+++
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/BenchmarkRunner.java
@@ -69,7 +69,7 @@ public class BenchmarkRunner {
executor.init();
BenchmarkMetrics metrics = new BenchmarkMetrics();
- MetricsCollector metricsCollector = new MetricsCollector(metrics);
+ MetricsCollector metricsCollector = new MetricsCollector(config,
metrics);
workloadGenerator = new WorkloadGenerator(config, executor,
metrics);
System.out.println("Starting benchmark...\n");
diff --git
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/config/BenchmarkConfig.java
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/config/BenchmarkConfig.java
index 8022b50c64..3966bdbb69 100644
---
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/config/BenchmarkConfig.java
+++
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/config/BenchmarkConfig.java
@@ -19,6 +19,8 @@ package org.apache.seata.benchmark.config;
import org.apache.seata.benchmark.constant.BenchmarkConstants;
import org.apache.seata.core.model.BranchType;
+import java.util.Locale;
+
/**
* Benchmark configuration
*/
@@ -34,6 +36,12 @@ public class BenchmarkConfig {
private String txServiceGroup = "default_tx_group";
private int rollbackPercentage = 2;
private int branches = 0;
+ private String sagaShape;
+ private String sagaWorkload = "mock";
+ private String sagaFailStep;
+ private Long sagaRandomSeed;
+ private String sagaTimeoutStep;
+ private int sagaTimeoutMs = 3000;
public BranchType getMode() {
return mode;
@@ -119,6 +127,54 @@ public class BenchmarkConfig {
this.branches = branches;
}
+ public String getSagaShape() {
+ return sagaShape;
+ }
+
+ public void setSagaShape(String sagaShape) {
+ this.sagaShape = sagaShape;
+ }
+
+ public String getSagaWorkload() {
+ return sagaWorkload;
+ }
+
+ public void setSagaWorkload(String sagaWorkload) {
+ this.sagaWorkload = sagaWorkload;
+ }
+
+ public String getSagaFailStep() {
+ return sagaFailStep;
+ }
+
+ public void setSagaFailStep(String sagaFailStep) {
+ this.sagaFailStep = sagaFailStep;
+ }
+
+ public Long getSagaRandomSeed() {
+ return sagaRandomSeed;
+ }
+
+ public void setSagaRandomSeed(Long sagaRandomSeed) {
+ this.sagaRandomSeed = sagaRandomSeed;
+ }
+
+ public String getSagaTimeoutStep() {
+ return sagaTimeoutStep;
+ }
+
+ public void setSagaTimeoutStep(String sagaTimeoutStep) {
+ this.sagaTimeoutStep = sagaTimeoutStep;
+ }
+
+ public int getSagaTimeoutMs() {
+ return sagaTimeoutMs;
+ }
+
+ public void setSagaTimeoutMs(int sagaTimeoutMs) {
+ this.sagaTimeoutMs = sagaTimeoutMs;
+ }
+
public void validate() {
validateMode();
validateNotEmpty(server, "server");
@@ -130,6 +186,11 @@ public class BenchmarkConfig {
validateNonNegative(warmupDuration, "warmupDuration");
validateNonNegative(branches, "branches");
validateRange(rollbackPercentage, 0, 100, "rollbackPercentage");
+ validateSagaShape();
+ validateSagaWorkload();
+ validateSagaFailStep();
+ validateSagaTimeoutStep();
+ validateNonNegative(sagaTimeoutMs, "sagaTimeoutMs");
validateTpsAndThreads();
}
@@ -163,6 +224,55 @@ public class BenchmarkConfig {
}
}
+ private void validateSagaShape() {
+ if (sagaShape == null || sagaShape.trim().isEmpty()) {
+ return;
+ }
+
+ String normalized = sagaShape.trim().toLowerCase(Locale.ROOT);
+ if (!"simple".equals(normalized) && !"order".equals(normalized)) {
+ throw new IllegalArgumentException("sagaShape must be one of:
simple, order");
+ }
+ sagaShape = normalized;
+ }
+
+ private void validateSagaWorkload() {
+ if (sagaWorkload == null || sagaWorkload.trim().isEmpty()) {
+ sagaWorkload = "mock";
+ return;
+ }
+
+ String normalized = sagaWorkload.trim().toLowerCase(Locale.ROOT);
+ if (!"mock".equals(normalized) && !"db".equals(normalized)) {
+ throw new IllegalArgumentException("sagaWorkload must be one of:
mock, db");
+ }
+ sagaWorkload = normalized;
+ }
+
+ private void validateSagaFailStep() {
+ if (sagaFailStep == null || sagaFailStep.trim().isEmpty()) {
+ return;
+ }
+
+ String normalized = sagaFailStep.trim().toLowerCase(Locale.ROOT);
+ if (!"inventory".equals(normalized) && !"payment".equals(normalized)
&& !"order".equals(normalized)) {
+ throw new IllegalArgumentException("sagaFailStep must be one of:
inventory, payment, order");
+ }
+ sagaFailStep = normalized;
+ }
+
+ private void validateSagaTimeoutStep() {
+ if (sagaTimeoutStep == null || sagaTimeoutStep.trim().isEmpty()) {
+ return;
+ }
+
+ String normalized = sagaTimeoutStep.trim().toLowerCase(Locale.ROOT);
+ if (!"inventory".equals(normalized) && !"payment".equals(normalized)
&& !"order".equals(normalized)) {
+ throw new IllegalArgumentException("sagaTimeoutStep must be one
of: inventory, payment, order");
+ }
+ sagaTimeoutStep = normalized;
+ }
+
private void validateTpsAndThreads() {
if (targetTps < BenchmarkConstants.UNLIMITED_TPS_THRESHOLD && threads
> 1) {
throw new IllegalArgumentException(String.format(
diff --git
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/config/BenchmarkConfigLoader.java
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/config/BenchmarkConfigLoader.java
index 14ccff8414..de62357590 100644
---
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/config/BenchmarkConfigLoader.java
+++
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/config/BenchmarkConfigLoader.java
@@ -99,7 +99,13 @@ public class BenchmarkConfigLoader {
String applicationId,
String txServiceGroup,
Integer rollbackPercentage,
- Integer branches) {
+ Integer branches,
+ String sagaShape,
+ String sagaWorkload,
+ String sagaFailStep,
+ Long sagaRandomSeed,
+ String sagaTimeoutStep,
+ Integer sagaTimeoutMs) {
if (StringUtils.isNotEmpty(server)) {
config.setServer(server);
}
@@ -130,6 +136,24 @@ public class BenchmarkConfigLoader {
if (branches != null && branches >= 0) {
config.setBranches(branches);
}
+ if (StringUtils.isNotEmpty(sagaShape)) {
+ config.setSagaShape(sagaShape);
+ }
+ if (StringUtils.isNotEmpty(sagaWorkload)) {
+ config.setSagaWorkload(sagaWorkload);
+ }
+ if (StringUtils.isNotEmpty(sagaFailStep)) {
+ config.setSagaFailStep(sagaFailStep);
+ }
+ if (sagaRandomSeed != null) {
+ config.setSagaRandomSeed(sagaRandomSeed);
+ }
+ if (StringUtils.isNotEmpty(sagaTimeoutStep)) {
+ config.setSagaTimeoutStep(sagaTimeoutStep);
+ }
+ if (sagaTimeoutMs != null && sagaTimeoutMs >= 0) {
+ config.setSagaTimeoutMs(sagaTimeoutMs);
+ }
return config;
}
}
diff --git
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/constant/BenchmarkConstants.java
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/constant/BenchmarkConstants.java
index c0445b691c..f536846ff5 100644
---
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/constant/BenchmarkConstants.java
+++
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/constant/BenchmarkConstants.java
@@ -41,6 +41,9 @@ public final class BenchmarkConstants {
public static final int INITIAL_BALANCE = 10000;
public static final int MIN_TRANSFER_AMOUNT = 1;
public static final int MAX_TRANSFER_AMOUNT = 100;
+ public static final int SAGA_PRODUCT_COUNT = 100;
+ public static final int SAGA_INITIAL_BALANCE = 1000000;
+ public static final int SAGA_INITIAL_INVENTORY = 100000;
// Thread pool configuration
public static final int SHUTDOWN_TIMEOUT_SECONDS = 10;
diff --git
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/executor/SagaModeExecutor.java
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/executor/SagaModeExecutor.java
index 2741d7a33c..75c962310c 100644
---
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/executor/SagaModeExecutor.java
+++
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/executor/SagaModeExecutor.java
@@ -19,9 +19,13 @@ package org.apache.seata.benchmark.executor;
import org.apache.seata.benchmark.config.BenchmarkConfig;
import org.apache.seata.benchmark.model.TransactionRecord;
import org.apache.seata.benchmark.saga.BenchmarkServiceInvoker;
+import org.apache.seata.benchmark.saga.InventoryDbSagaService;
import org.apache.seata.benchmark.saga.InventorySagaService;
+import org.apache.seata.benchmark.saga.OrderDbSagaService;
import org.apache.seata.benchmark.saga.OrderSagaService;
+import org.apache.seata.benchmark.saga.PaymentDbSagaService;
import org.apache.seata.benchmark.saga.PaymentSagaService;
+import org.apache.seata.benchmark.saga.SagaDbEnvironment;
import org.apache.seata.benchmark.saga.SimpleSpelExpressionFactory;
import org.apache.seata.core.exception.TransactionException;
import org.apache.seata.core.model.GlobalStatus;
@@ -65,6 +69,12 @@ public class SagaModeExecutor implements TransactionExecutor
{
private static final String SIMPLE_SAGA_NAME = "benchmarkSimpleSaga";
private static final String ORDER_SAGA_NAME = "benchmarkOrderSaga";
+ private static final String STEP_INVENTORY = "inventory";
+ private static final String STEP_PAYMENT = "payment";
+ private static final String STEP_ORDER = "order";
+ private static final String SHAPE_SIMPLE = "simple";
+ private static final String SHAPE_ORDER = "order";
+ private static final String WORKLOAD_DB = "db";
private final BenchmarkConfig config;
private StateMachineEngine stateMachineEngine;
@@ -95,6 +105,12 @@ public class SagaModeExecutor implements
TransactionExecutor {
// Create and configure state machine config
stateMachineConfig = new BenchmarkStateMachineConfig();
stateMachineConfig.setRollbackPercentage(config.getRollbackPercentage());
+ stateMachineConfig.setSagaFailStep(config.getSagaFailStep());
+ stateMachineConfig.setSagaRandomSeed(config.getSagaRandomSeed());
+ stateMachineConfig.setSagaTimeoutStep(config.getSagaTimeoutStep());
+ stateMachineConfig.setSagaTimeoutMs(config.getSagaTimeoutMs());
+ stateMachineConfig.setSagaWorkload(config.getSagaWorkload());
+ stateMachineConfig.setBenchmarkConfig(config);
stateMachineConfig.init();
// Create state machine engine
@@ -106,6 +122,9 @@ public class SagaModeExecutor implements
TransactionExecutor {
LOGGER.info("Available state machines: {}, {}", SIMPLE_SAGA_NAME,
ORDER_SAGA_NAME);
} catch (Exception e) {
+ if (stateMachineConfig != null) {
+ stateMachineConfig.destroy();
+ }
throw new RuntimeException("Failed to initialize Saga state
machine engine", e);
}
}
@@ -173,8 +192,7 @@ public class SagaModeExecutor implements
TransactionExecutor {
boolean success = false;
try {
- // Choose state machine based on branch count
- String stateMachineName = branchCount >= 3 ? ORDER_SAGA_NAME :
SIMPLE_SAGA_NAME;
+ String stateMachineName = resolveStateMachineName(branchCount);
// Prepare start parameters
Map<String, Object> startParams = createStartParams();
@@ -187,19 +205,17 @@ public class SagaModeExecutor implements
TransactionExecutor {
ExecutionStatus executionStatus = instance.getStatus();
ExecutionStatus compensationStatus =
instance.getCompensationStatus();
- if (ExecutionStatus.SU.equals(executionStatus)) {
+ if (ExecutionStatus.SU.equals(compensationStatus)) {
+ status = STATUS_COMPENSATED;
+ } else if (ExecutionStatus.FA.equals(compensationStatus)) {
+ status = STATUS_COMPENSATION_FAILED;
+ } else if (ExecutionStatus.UN.equals(compensationStatus)) {
+ status = STATUS_UNKNOWN;
+ } else if (ExecutionStatus.SU.equals(executionStatus)) {
status = STATUS_COMMITTED;
success = true;
} else if (ExecutionStatus.FA.equals(executionStatus)) {
- if (compensationStatus != null) {
- if (ExecutionStatus.SU.equals(compensationStatus)) {
- status = STATUS_COMPENSATED;
- } else {
- status = STATUS_COMPENSATION_FAILED;
- }
- } else {
- status = STATUS_FAILED;
- }
+ status = STATUS_FAILED;
} else if (ExecutionStatus.UN.equals(executionStatus)) {
status = STATUS_UNKNOWN;
} else {
@@ -215,6 +231,16 @@ public class SagaModeExecutor implements
TransactionExecutor {
return new TransactionRecord(businessKey, status, duration,
branchCount, success);
}
+ private String resolveStateMachineName(int branchCount) {
+ if (SHAPE_ORDER.equals(config.getSagaShape())) {
+ return ORDER_SAGA_NAME;
+ }
+ if (SHAPE_SIMPLE.equals(config.getSagaShape())) {
+ return SIMPLE_SAGA_NAME;
+ }
+ return branchCount >= 3 ? ORDER_SAGA_NAME : SIMPLE_SAGA_NAME;
+ }
+
private Map<String, Object> createStartParams() {
Map<String, Object> params = new HashMap<>();
params.put("userId", "user-" +
ThreadLocalRandom.current().nextInt(1000));
@@ -251,7 +277,9 @@ public class SagaModeExecutor implements
TransactionExecutor {
private void destroyRealMode() {
LOGGER.info("Destroying Real Saga mode resources");
- // StateMachineEngine doesn't have a close method
+ if (stateMachineConfig != null) {
+ stateMachineConfig.destroy();
+ }
stateMachineEngine = null;
stateMachineConfig = null;
}
@@ -262,11 +290,48 @@ public class SagaModeExecutor implements
TransactionExecutor {
private static class BenchmarkStateMachineConfig extends
AbstractStateMachineConfig {
private int rollbackPercentage = 0;
+ private String sagaFailStep;
+ private Long sagaRandomSeed;
+ private String sagaTimeoutStep;
+ private int sagaTimeoutMs = 3000;
+ private String sagaWorkload = "mock";
+ private BenchmarkConfig benchmarkConfig;
+ private SagaDbEnvironment sagaDbEnvironment;
+ private OrderSagaService orderSagaService;
+ private InventorySagaService inventorySagaService;
+ private PaymentSagaService paymentSagaService;
+ private OrderDbSagaService orderDbSagaService;
+ private InventoryDbSagaService inventoryDbSagaService;
+ private PaymentDbSagaService paymentDbSagaService;
public void setRollbackPercentage(int rollbackPercentage) {
this.rollbackPercentage = rollbackPercentage;
}
+ public void setSagaFailStep(String sagaFailStep) {
+ this.sagaFailStep = sagaFailStep;
+ }
+
+ public void setSagaRandomSeed(Long sagaRandomSeed) {
+ this.sagaRandomSeed = sagaRandomSeed;
+ }
+
+ public void setSagaTimeoutStep(String sagaTimeoutStep) {
+ this.sagaTimeoutStep = sagaTimeoutStep;
+ }
+
+ public void setSagaTimeoutMs(int sagaTimeoutMs) {
+ this.sagaTimeoutMs = sagaTimeoutMs;
+ }
+
+ public void setSagaWorkload(String sagaWorkload) {
+ this.sagaWorkload = sagaWorkload;
+ }
+
+ public void setBenchmarkConfig(BenchmarkConfig benchmarkConfig) {
+ this.benchmarkConfig = benchmarkConfig;
+ }
+
@Override
public void init() throws Exception {
// Load state machine definitions from classpath
@@ -304,13 +369,41 @@ public class SagaModeExecutor implements
TransactionExecutor {
// Register benchmark services with the service invoker manager
BenchmarkServiceInvoker serviceInvoker = new
BenchmarkServiceInvoker();
- // Register services with configured rollback percentage
- // Divide rollback percentage by 3 for each service so total
probability is approximately correct
- int serviceRollbackPct = rollbackPercentage > 0 ? Math.max(1,
rollbackPercentage / 3) : 0;
-
- serviceInvoker.registerService("orderService", new
OrderSagaService(serviceRollbackPct, 5));
- serviceInvoker.registerService("inventoryService", new
InventorySagaService(serviceRollbackPct, 5));
- serviceInvoker.registerService("paymentService", new
PaymentSagaService(serviceRollbackPct, 5));
+ boolean restrictFailureStep =
+ sagaFailStep != null && !sagaFailStep.trim().isEmpty();
+ int serviceRollbackPct = restrictFailureStep
+ ? rollbackPercentage
+ : (rollbackPercentage > 0 ? Math.max(1,
rollbackPercentage / 3) : 0);
+
+ boolean orderFailEnabled = !restrictFailureStep ||
STEP_ORDER.equals(sagaFailStep);
+ boolean inventoryFailEnabled = !restrictFailureStep ||
STEP_INVENTORY.equals(sagaFailStep);
+ boolean paymentFailEnabled = !restrictFailureStep ||
STEP_PAYMENT.equals(sagaFailStep);
+ boolean orderTimeoutEnabled =
STEP_ORDER.equals(sagaTimeoutStep);
+ boolean inventoryTimeoutEnabled =
STEP_INVENTORY.equals(sagaTimeoutStep);
+ boolean paymentTimeoutEnabled =
STEP_PAYMENT.equals(sagaTimeoutStep);
+
+ if (WORKLOAD_DB.equals(sagaWorkload)) {
+ initDbEnvironment();
+ registerDbServices(
+ serviceInvoker,
+ serviceRollbackPct,
+ orderFailEnabled,
+ inventoryFailEnabled,
+ paymentFailEnabled,
+ orderTimeoutEnabled,
+ inventoryTimeoutEnabled,
+ paymentTimeoutEnabled);
+ } else {
+ registerMockServices(
+ serviceInvoker,
+ serviceRollbackPct,
+ orderFailEnabled,
+ inventoryFailEnabled,
+ paymentFailEnabled,
+ orderTimeoutEnabled,
+ inventoryTimeoutEnabled,
+ paymentTimeoutEnabled);
+ }
// Register the service invoker for different service types
getServiceInvokerManager().putServiceInvoker(DomainConstants.SERVICE_TYPE_SPRING_BEAN,
serviceInvoker);
@@ -326,5 +419,126 @@ public class SagaModeExecutor implements
TransactionExecutor {
}
return buffer.toByteArray();
}
+
+ private ThreadLocal<java.util.Random> createFailureRandom(int salt) {
+ return
org.apache.seata.benchmark.saga.FailureRandomProvider.create(
+ sagaRandomSeed == null ? null : sagaRandomSeed + salt);
+ }
+
+ private void registerMockServices(
+ BenchmarkServiceInvoker serviceInvoker,
+ int serviceRollbackPct,
+ boolean orderFailEnabled,
+ boolean inventoryFailEnabled,
+ boolean paymentFailEnabled,
+ boolean orderTimeoutEnabled,
+ boolean inventoryTimeoutEnabled,
+ boolean paymentTimeoutEnabled) {
+ orderSagaService = new OrderSagaService(
+ serviceRollbackPct,
+ 5,
+ orderFailEnabled,
+ createFailureRandom(11),
+ orderTimeoutEnabled,
+ sagaTimeoutMs);
+ serviceInvoker.registerService("orderService", orderSagaService);
+ inventorySagaService = new InventorySagaService(
+ serviceRollbackPct,
+ 5,
+ inventoryFailEnabled,
+ createFailureRandom(17),
+ inventoryTimeoutEnabled,
+ sagaTimeoutMs);
+ serviceInvoker.registerService("inventoryService",
inventorySagaService);
+ paymentSagaService = new PaymentSagaService(
+ serviceRollbackPct,
+ 5,
+ paymentFailEnabled,
+ createFailureRandom(23),
+ paymentTimeoutEnabled,
+ sagaTimeoutMs);
+ serviceInvoker.registerService("paymentService",
paymentSagaService);
+ }
+
+ private void registerDbServices(
+ BenchmarkServiceInvoker serviceInvoker,
+ int serviceRollbackPct,
+ boolean orderFailEnabled,
+ boolean inventoryFailEnabled,
+ boolean paymentFailEnabled,
+ boolean orderTimeoutEnabled,
+ boolean inventoryTimeoutEnabled,
+ boolean paymentTimeoutEnabled) {
+ orderDbSagaService = new OrderDbSagaService(
+ sagaDbEnvironment.getDataSource(),
+ serviceRollbackPct,
+ 5,
+ orderFailEnabled,
+ createFailureRandom(11),
+ orderTimeoutEnabled,
+ sagaTimeoutMs);
+ serviceInvoker.registerService("orderService", orderDbSagaService);
+ inventoryDbSagaService = new InventoryDbSagaService(
+ sagaDbEnvironment.getDataSource(),
+ serviceRollbackPct,
+ 5,
+ inventoryFailEnabled,
+ createFailureRandom(17),
+ inventoryTimeoutEnabled,
+ sagaTimeoutMs);
+ serviceInvoker.registerService("inventoryService",
inventoryDbSagaService);
+ paymentDbSagaService = new PaymentDbSagaService(
+ sagaDbEnvironment.getDataSource(),
+ serviceRollbackPct,
+ 5,
+ paymentFailEnabled,
+ createFailureRandom(23),
+ paymentTimeoutEnabled,
+ sagaTimeoutMs);
+ serviceInvoker.registerService("paymentService",
paymentDbSagaService);
+ }
+
+ private void initDbEnvironment() {
+ if (benchmarkConfig == null) {
+ throw new IllegalStateException("BenchmarkConfig is required
for DB-backed Saga workload");
+ }
+ sagaDbEnvironment = new SagaDbEnvironment(benchmarkConfig);
+ sagaDbEnvironment.init();
+ }
+
+ public void destroy() {
+ destroyServices();
+ if (sagaDbEnvironment != null) {
+ sagaDbEnvironment.destroy();
+ sagaDbEnvironment = null;
+ }
+ }
+
+ private void destroyServices() {
+ if (orderSagaService != null) {
+ orderSagaService.destroy();
+ orderSagaService = null;
+ }
+ if (inventorySagaService != null) {
+ inventorySagaService.destroy();
+ inventorySagaService = null;
+ }
+ if (paymentSagaService != null) {
+ paymentSagaService.destroy();
+ paymentSagaService = null;
+ }
+ if (orderDbSagaService != null) {
+ orderDbSagaService.destroy();
+ orderDbSagaService = null;
+ }
+ if (inventoryDbSagaService != null) {
+ inventoryDbSagaService.destroy();
+ inventoryDbSagaService = null;
+ }
+ if (paymentDbSagaService != null) {
+ paymentDbSagaService.destroy();
+ paymentDbSagaService = null;
+ }
+ }
}
}
diff --git
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/executor/WorkloadGenerator.java
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/executor/WorkloadGenerator.java
index 25eccb0a62..d26b99ccc8 100644
---
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/executor/WorkloadGenerator.java
+++
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/executor/WorkloadGenerator.java
@@ -34,6 +34,8 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import static
org.apache.seata.benchmark.constant.BenchmarkConstants.STATUS_FAILED;
+
/**
* Workload generator with TPS rate limiting
*/
@@ -100,20 +102,17 @@ public class WorkloadGenerator {
}
private void executeTransaction() {
+ long startTime = System.currentTimeMillis();
try {
TransactionRecord record = executor.execute();
-
- if (record.isSuccess()) {
- metrics.recordSuccess(record.getDuration());
- } else {
- metrics.recordFailure(record.getDuration());
- }
+ metrics.recordTransaction(record.getStatus(),
record.getDuration());
addRecentRecord(record);
} catch (Exception e) {
LOGGER.error("Transaction execution error", e);
- metrics.recordFailure(0);
+ long duration = System.currentTimeMillis() - startTime;
+ metrics.recordTransaction(STATUS_FAILED, duration);
}
}
diff --git
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/model/BenchmarkMetrics.java
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/model/BenchmarkMetrics.java
index bc130823cd..58ad94a38b 100644
---
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/model/BenchmarkMetrics.java
+++
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/model/BenchmarkMetrics.java
@@ -32,10 +32,14 @@ public class BenchmarkMetrics {
private final AtomicLong totalCount = new AtomicLong(0);
private final AtomicLong successCount = new AtomicLong(0);
private final AtomicLong failedCount = new AtomicLong(0);
+ private final AtomicLong committedCount = new AtomicLong(0);
+ private final AtomicLong compensatedCount = new AtomicLong(0);
+ private final AtomicLong compensationFailedCount = new AtomicLong(0);
+ private final AtomicLong unknownCount = new AtomicLong(0);
private final List<Long> latencies =
Collections.synchronizedList(new
ArrayList<>(BenchmarkConstants.MAX_LATENCY_SAMPLES));
private final AtomicLong totalSamples = new AtomicLong(0);
- private final long startTime = System.currentTimeMillis();
+ private volatile long startTime = System.currentTimeMillis();
private volatile long lastCountSnapshot = 0;
private volatile long lastSnapshotTime = System.currentTimeMillis();
@@ -55,6 +59,29 @@ public class BenchmarkMetrics {
addLatencySample(latencyMs);
}
+ public void recordTransaction(String status, long latencyMs) {
+ totalCount.incrementAndGet();
+
+ if (BenchmarkConstants.STATUS_COMMITTED.equals(status)) {
+ successCount.incrementAndGet();
+ committedCount.incrementAndGet();
+ } else if (BenchmarkConstants.STATUS_COMPENSATED.equals(status)) {
+ compensatedCount.incrementAndGet();
+ successCount.incrementAndGet();
+ } else if
(BenchmarkConstants.STATUS_COMPENSATION_FAILED.equals(status)) {
+ compensationFailedCount.incrementAndGet();
+ failedCount.incrementAndGet();
+ } else if (BenchmarkConstants.STATUS_FAILED.equals(status)
+ || BenchmarkConstants.STATUS_ROLLBACKED.equals(status)) {
+ failedCount.incrementAndGet();
+ } else {
+ unknownCount.incrementAndGet();
+ failedCount.incrementAndGet();
+ }
+
+ addLatencySample(latencyMs);
+ }
+
private void addLatencySample(long latencyMs) {
totalSamples.incrementAndGet();
synchronized (latencies) {
@@ -80,6 +107,26 @@ public class BenchmarkMetrics {
return failedCount.get();
}
+ public long getCommittedCount() {
+ return committedCount.get();
+ }
+
+ public long getCompensatedCount() {
+ return compensatedCount.get();
+ }
+
+ public long getExecutionFailedCount() {
+ return failedCount.get() - compensationFailedCount.get() -
unknownCount.get();
+ }
+
+ public long getCompensationFailedCount() {
+ return compensationFailedCount.get();
+ }
+
+ public long getUnknownCount() {
+ return unknownCount.get();
+ }
+
public double getSuccessRate() {
long total = totalCount.get();
if (total == 0) {
@@ -88,6 +135,30 @@ public class BenchmarkMetrics {
return (double) successCount.get() / total * 100;
}
+ public double getCommittedRate() {
+ long total = totalCount.get();
+ if (total == 0) {
+ return 0.0;
+ }
+ return (double) committedCount.get() / total * 100;
+ }
+
+ public double getCompensatedRate() {
+ long total = totalCount.get();
+ if (total == 0) {
+ return 0.0;
+ }
+ return (double) compensatedCount.get() / total * 100;
+ }
+
+ public double getEndStateSuccessRate() {
+ long total = totalCount.get();
+ if (total == 0) {
+ return 0.0;
+ }
+ return (double) (committedCount.get() + compensatedCount.get()) /
total * 100;
+ }
+
public double getAverageTps() {
long elapsed = System.currentTimeMillis() - startTime;
if (elapsed == 0) {
@@ -149,12 +220,20 @@ public class BenchmarkMetrics {
}
public void reset() {
+ startTime = System.currentTimeMillis();
totalCount.set(0);
successCount.set(0);
failedCount.set(0);
+ committedCount.set(0);
+ compensatedCount.set(0);
+ compensationFailedCount.set(0);
+ unknownCount.set(0);
+ totalSamples.set(0);
latencies.clear();
lastCountSnapshot = 0;
lastSnapshotTime = System.currentTimeMillis();
+ cachedStats = null;
+ lastStatsUpdateTime = 0;
}
public static class LatencyStats {
diff --git
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/monitor/MetricsCollector.java
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/monitor/MetricsCollector.java
index f784b1d884..84a4221f27 100644
---
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/monitor/MetricsCollector.java
+++
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/monitor/MetricsCollector.java
@@ -16,7 +16,9 @@
*/
package org.apache.seata.benchmark.monitor;
+import org.apache.seata.benchmark.config.BenchmarkConfig;
import org.apache.seata.benchmark.model.BenchmarkMetrics;
+import org.apache.seata.core.model.BranchType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -33,19 +35,38 @@ public class MetricsCollector {
private static final Logger LOGGER =
LoggerFactory.getLogger(MetricsCollector.class);
+ private final BenchmarkConfig config;
private final BenchmarkMetrics metrics;
- public MetricsCollector(BenchmarkMetrics metrics) {
+ public MetricsCollector(BenchmarkConfig config, BenchmarkMetrics metrics) {
+ this.config = config;
this.metrics = metrics;
}
public void exportToCsv(String filename) {
try (PrintWriter writer = new PrintWriter(new FileWriter(filename))) {
writer.println("Metric,Value");
+ writer.println("Mode," + config.getMode());
+ if (config.getMode() == BranchType.SAGA
+ && config.getSagaWorkload() != null
+ && !config.getSagaWorkload().isEmpty()) {
+ writer.println("Saga Workload," + config.getSagaWorkload());
+ }
+ if (config.getSagaShape() != null &&
!config.getSagaShape().isEmpty()) {
+ writer.println("Saga Shape," + config.getSagaShape());
+ }
writer.println("Total Transactions," + metrics.getTotalCount());
writer.println("Success Count," + metrics.getSuccessCount());
writer.println("Failed Count," + metrics.getFailedCount());
+ writer.println("Committed Count," + metrics.getCommittedCount());
+ writer.println("Compensated Count," +
metrics.getCompensatedCount());
+ writer.println("Execution Failed Count," +
metrics.getExecutionFailedCount());
+ writer.println("Compensation Failed Count," +
metrics.getCompensationFailedCount());
+ writer.println("Unknown Count," + metrics.getUnknownCount());
writer.println("Success Rate (%)," + String.format("%.2f",
metrics.getSuccessRate()));
+ writer.println("Committed Rate (%)," + String.format("%.2f",
metrics.getCommittedRate()));
+ writer.println("Compensated Rate (%)," + String.format("%.2f",
metrics.getCompensatedRate()));
+ writer.println("End-State Success Rate (%)," +
String.format("%.2f", metrics.getEndStateSuccessRate()));
writer.println("Average TPS," + String.format("%.2f",
metrics.getAverageTps()));
writer.println("Elapsed Time (s)," +
metrics.getElapsedTimeSeconds());
@@ -71,10 +92,27 @@ public class MetricsCollector {
report.append("===================================================\n");
report.append(" Seata Benchmark Final Report\n");
report.append("===================================================\n");
+ report.append(String.format("Mode: %s\n",
config.getMode()));
+ if (config.getMode() == BranchType.SAGA
+ && config.getSagaWorkload() != null
+ && !config.getSagaWorkload().isEmpty()) {
+ report.append(String.format("Saga Workload: %s\n",
config.getSagaWorkload()));
+ }
+ if (config.getSagaShape() != null && !config.getSagaShape().isEmpty())
{
+ report.append(String.format("Saga Shape: %s\n",
config.getSagaShape()));
+ }
report.append(String.format("Total Transactions: %,d\n",
metrics.getTotalCount()));
report.append(String.format("Success Count: %,d\n",
metrics.getSuccessCount()));
report.append(String.format("Failed Count: %,d\n",
metrics.getFailedCount()));
report.append(String.format("Success Rate: %.2f%%\n",
metrics.getSuccessRate()));
+ report.append(String.format("Committed Count: %,d\n",
metrics.getCommittedCount()));
+ report.append(String.format("Compensated Count: %,d\n",
metrics.getCompensatedCount()));
+ report.append(String.format("Execution Failed Count: %,d\n",
metrics.getExecutionFailedCount()));
+ report.append(String.format("Compensation Failed Count: %,d\n",
metrics.getCompensationFailedCount()));
+ report.append(String.format("Unknown Count: %,d\n",
metrics.getUnknownCount()));
+ report.append(String.format("Committed Rate: %.2f%%\n",
metrics.getCommittedRate()));
+ report.append(String.format("Compensated Rate: %.2f%%\n",
metrics.getCompensatedRate()));
+ report.append(String.format("End-State Success Rate: %.2f%%\n",
metrics.getEndStateSuccessRate()));
report.append(String.format("Average TPS: %.2f\n",
metrics.getAverageTps()));
report.append(String.format("Elapsed Time: %d seconds\n",
metrics.getElapsedTimeSeconds()));
report.append("\n");
diff --git
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/FailureRandomProvider.java
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/FailureRandomProvider.java
new file mode 100644
index 0000000000..7375812d7c
--- /dev/null
+++
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/FailureRandomProvider.java
@@ -0,0 +1,51 @@
+/*
+ * 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.seata.benchmark.saga;
+
+import java.util.Random;
+import java.util.concurrent.ThreadLocalRandom;
+
+/**
+ * Thread-local failure random provider to avoid cross-thread contention while
keeping seeded behavior reproducible.
+ */
+public final class FailureRandomProvider {
+
+ private FailureRandomProvider() {}
+
+ public static ThreadLocal<Random> create(Long seed) {
+ if (seed == null) {
+ return null;
+ }
+ return ThreadLocal.withInitial(
+ () -> new Random(mixSeed(seed,
Thread.currentThread().getName())));
+ }
+
+ public static int nextPercent(ThreadLocal<Random> failureRandom) {
+ if (failureRandom != null) {
+ return failureRandom.get().nextInt(100);
+ }
+ return ThreadLocalRandom.current().nextInt(100);
+ }
+
+ private static long mixSeed(long seed, String threadName) {
+ long mixed = seed ^ threadName.hashCode();
+ mixed ^= (mixed << 21);
+ mixed ^= (mixed >>> 35);
+ mixed ^= (mixed << 4);
+ return mixed;
+ }
+}
diff --git
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/InventoryDbSagaService.java
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/InventoryDbSagaService.java
new file mode 100644
index 0000000000..cbdc66d39d
--- /dev/null
+++
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/InventoryDbSagaService.java
@@ -0,0 +1,183 @@
+/*
+ * 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.seata.benchmark.saga;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Random;
+import java.util.concurrent.ThreadLocalRandom;
+
+/**
+ * Database-backed inventory service for Saga benchmark.
+ * @author zihenzzz
+ */
+public class InventoryDbSagaService {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(InventoryDbSagaService.class);
+
+ private final DataSource dataSource;
+ private final int rollbackPercentage;
+ private final int simulatedDelayMs;
+ private final boolean failInjectionEnabled;
+ private final ThreadLocal<Random> failureRandom;
+ private final boolean timeoutInjectionEnabled;
+ private final int timeoutMs;
+
+ public InventoryDbSagaService(
+ DataSource dataSource,
+ int rollbackPercentage,
+ int simulatedDelayMs,
+ boolean failInjectionEnabled,
+ ThreadLocal<Random> failureRandom,
+ boolean timeoutInjectionEnabled,
+ int timeoutMs) {
+ this.dataSource = dataSource;
+ this.rollbackPercentage = rollbackPercentage;
+ this.simulatedDelayMs = simulatedDelayMs;
+ this.failInjectionEnabled = failInjectionEnabled;
+ this.failureRandom = failureRandom;
+ this.timeoutInjectionEnabled = timeoutInjectionEnabled;
+ this.timeoutMs = timeoutMs;
+ }
+
+ public Map<String, Object> reserveInventory(Map<String, Object> params) {
+ String productId = (String) params.get("productId");
+ Integer quantity = (Integer) params.get("quantity");
+
+ simulateDelay();
+ if (timeoutInjectionEnabled) {
+ simulateTimeout("inventory reservation");
+ }
+ if (shouldFail()) {
+ throw new RuntimeException("Simulated inventory reservation
failure");
+ }
+
+ try (Connection conn = dataSource.getConnection()) {
+ boolean originalAutoCommit = conn.getAutoCommit();
+ conn.setAutoCommit(false);
+ try {
+ try (PreparedStatement pstmt = conn.prepareStatement("UPDATE
benchmark_inventory "
+ + "SET available_qty = available_qty - ?, reserved_qty
= reserved_qty + ? "
+ + "WHERE product_id = ? AND available_qty >= ?")) {
+ pstmt.setInt(1, quantity);
+ pstmt.setInt(2, quantity);
+ pstmt.setString(3, productId);
+ pstmt.setInt(4, quantity);
+ if (pstmt.executeUpdate() == 0) {
+ throw new RuntimeException("Insufficient inventory for
product " + productId);
+ }
+ }
+ conn.commit();
+ } catch (Exception e) {
+ rollbackQuietly(conn, e);
+ throw e;
+ } finally {
+ restoreAutoCommit(conn, originalAutoCommit);
+ }
+ } catch (SQLException e) {
+ throw new RuntimeException("Failed to reserve inventory", e);
+ }
+
+ LOGGER.debug("Reserved inventory in DB: productId={}, quantity={}",
productId, quantity);
+ Map<String, Object> result = new HashMap<>();
+ result.put("code", "S");
+ result.put("productId", productId);
+ result.put("quantity", quantity);
+ return result;
+ }
+
+ public Map<String, Object> releaseInventory(Map<String, Object> params) {
+ String productId = (String) params.get("productId");
+ Integer quantity = (Integer) params.get("quantity");
+
+ simulateDelay();
+
+ try (Connection conn = dataSource.getConnection();
+ PreparedStatement pstmt = conn.prepareStatement("UPDATE
benchmark_inventory "
+ + "SET available_qty = available_qty + ?, reserved_qty
= GREATEST(reserved_qty - ?, 0) "
+ + "WHERE product_id = ?")) {
+ pstmt.setInt(1, quantity);
+ pstmt.setInt(2, quantity);
+ pstmt.setString(3, productId);
+ pstmt.executeUpdate();
+ } catch (SQLException e) {
+ throw new RuntimeException("Failed to release inventory", e);
+ }
+
+ Map<String, Object> result = new HashMap<>();
+ result.put("code", "S");
+ result.put("productId", productId);
+ result.put("quantity", quantity);
+ return result;
+ }
+
+ public void destroy() {
+ if (failureRandom != null) {
+ failureRandom.remove();
+ }
+ }
+
+ private void rollbackQuietly(Connection conn, Exception original) {
+ try {
+ conn.rollback();
+ } catch (SQLException rollbackException) {
+ original.addSuppressed(rollbackException);
+ }
+ }
+
+ private void restoreAutoCommit(Connection conn, boolean
originalAutoCommit) {
+ try {
+ conn.setAutoCommit(originalAutoCommit);
+ } catch (SQLException e) {
+ throw new RuntimeException("Failed to restore auto-commit for
inventory connection", e);
+ }
+ }
+
+ private void simulateDelay() {
+ if (simulatedDelayMs > 0) {
+ try {
+
Thread.sleep(ThreadLocalRandom.current().nextInt(simulatedDelayMs));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
+ private boolean shouldFail() {
+ return failInjectionEnabled && nextFailurePercent() <
rollbackPercentage;
+ }
+
+ private void simulateTimeout(String operation) {
+ try {
+ Thread.sleep(timeoutMs);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ throw new RuntimeException("Simulated " + operation + " timeout");
+ }
+
+ private int nextFailurePercent() {
+ return FailureRandomProvider.nextPercent(failureRandom);
+ }
+}
diff --git
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/InventorySagaService.java
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/InventorySagaService.java
index d75bc53eca..3e330216a7 100644
---
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/InventorySagaService.java
+++
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/InventorySagaService.java
@@ -21,6 +21,7 @@ import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.Map;
+import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;
/**
@@ -33,10 +34,24 @@ public class InventorySagaService {
private final int rollbackPercentage;
private final int simulatedDelayMs;
-
- public InventorySagaService(int rollbackPercentage, int simulatedDelayMs) {
+ private final boolean failInjectionEnabled;
+ private final ThreadLocal<Random> failureRandom;
+ private final boolean timeoutInjectionEnabled;
+ private final int timeoutMs;
+
+ public InventorySagaService(
+ int rollbackPercentage,
+ int simulatedDelayMs,
+ boolean failInjectionEnabled,
+ ThreadLocal<Random> failureRandom,
+ boolean timeoutInjectionEnabled,
+ int timeoutMs) {
this.rollbackPercentage = rollbackPercentage;
this.simulatedDelayMs = simulatedDelayMs;
+ this.failInjectionEnabled = failInjectionEnabled;
+ this.failureRandom = failureRandom;
+ this.timeoutInjectionEnabled = timeoutInjectionEnabled;
+ this.timeoutMs = timeoutMs;
}
/**
@@ -54,7 +69,11 @@ public class InventorySagaService {
// Simulate processing time
simulateDelay();
- // Simulate random failure based on rollback percentage
+ if (timeoutInjectionEnabled) {
+ simulateTimeout("inventory reservation");
+ }
+
+ // Simulate failure injection on the selected step.
if (shouldFail()) {
LOGGER.debug("Inventory reservation failed (simulated):
productId={}", productId);
throw new RuntimeException("Simulated inventory reservation
failure");
@@ -91,6 +110,12 @@ public class InventorySagaService {
return result;
}
+ public void destroy() {
+ if (failureRandom != null) {
+ failureRandom.remove();
+ }
+ }
+
private void simulateDelay() {
if (simulatedDelayMs > 0) {
try {
@@ -102,6 +127,23 @@ public class InventorySagaService {
}
private boolean shouldFail() {
- return ThreadLocalRandom.current().nextInt(100) < rollbackPercentage;
+ if (!failInjectionEnabled) {
+ return false;
+ }
+ return nextFailurePercent() < rollbackPercentage;
+ }
+
+ private void simulateTimeout(String operation) {
+ LOGGER.debug("Simulating {} timeout: {} ms", operation, timeoutMs);
+ try {
+ Thread.sleep(timeoutMs);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ throw new RuntimeException("Simulated " + operation + " timeout");
+ }
+
+ private int nextFailurePercent() {
+ return FailureRandomProvider.nextPercent(failureRandom);
}
}
diff --git
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/OrderDbSagaService.java
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/OrderDbSagaService.java
new file mode 100644
index 0000000000..4d4715e695
--- /dev/null
+++
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/OrderDbSagaService.java
@@ -0,0 +1,158 @@
+/*
+ * 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.seata.benchmark.saga;
+
+import javax.sql.DataSource;
+import java.math.BigDecimal;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Random;
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+
+/**
+ * Database-backed order service for Saga benchmark.
+ * @author zihenzzz
+ */
+public class OrderDbSagaService {
+
+ private final DataSource dataSource;
+ private final int rollbackPercentage;
+ private final int simulatedDelayMs;
+ private final boolean failInjectionEnabled;
+ private final ThreadLocal<Random> failureRandom;
+ private final boolean timeoutInjectionEnabled;
+ private final int timeoutMs;
+
+ public OrderDbSagaService(
+ DataSource dataSource,
+ int rollbackPercentage,
+ int simulatedDelayMs,
+ boolean failInjectionEnabled,
+ ThreadLocal<Random> failureRandom,
+ boolean timeoutInjectionEnabled,
+ int timeoutMs) {
+ this.dataSource = dataSource;
+ this.rollbackPercentage = rollbackPercentage;
+ this.simulatedDelayMs = simulatedDelayMs;
+ this.failInjectionEnabled = failInjectionEnabled;
+ this.failureRandom = failureRandom;
+ this.timeoutInjectionEnabled = timeoutInjectionEnabled;
+ this.timeoutMs = timeoutMs;
+ }
+
+ public Map<String, Object> createOrder(Map<String, Object> params) {
+ String orderId = UUID.randomUUID().toString().substring(0, 8);
+ String userId = (String) params.get("userId");
+ String productId = (String) params.get("productId");
+ Integer quantity = (Integer) params.get("quantity");
+ BigDecimal amount = toBigDecimal(params.get("amount"));
+
+ simulateDelay();
+ if (timeoutInjectionEnabled) {
+ simulateTimeout("order creation");
+ }
+ if (shouldFail()) {
+ throw new RuntimeException("Simulated order creation failure");
+ }
+
+ try (Connection conn = dataSource.getConnection();
+ PreparedStatement pstmt = conn.prepareStatement("INSERT INTO
benchmark_order "
+ + "(order_id, user_id, product_id, quantity, amount,
status) "
+ + "VALUES (?, ?, ?, ?, ?, ?)")) {
+ pstmt.setString(1, orderId);
+ pstmt.setString(2, userId);
+ pstmt.setString(3, productId);
+ pstmt.setInt(4, quantity);
+ pstmt.setBigDecimal(5, amount);
+ pstmt.setString(6, "CREATED");
+ pstmt.executeUpdate();
+ } catch (SQLException e) {
+ throw new RuntimeException("Failed to create order", e);
+ }
+
+ Map<String, Object> result = new HashMap<>();
+ result.put("code", "S");
+ result.put("orderId", orderId);
+ result.put("userId", userId);
+ result.put("productId", productId);
+ return result;
+ }
+
+ public Map<String, Object> cancelOrder(Map<String, Object> params) {
+ String orderId = (String) params.get("orderId");
+ String userId = (String) params.get("userId");
+
+ simulateDelay();
+
+ try (Connection conn = dataSource.getConnection();
+ PreparedStatement pstmt =
+ conn.prepareStatement("UPDATE benchmark_order SET
status = ? WHERE order_id = ?")) {
+ pstmt.setString(1, "CANCELLED");
+ pstmt.setString(2, orderId);
+ pstmt.executeUpdate();
+ } catch (SQLException e) {
+ throw new RuntimeException("Failed to cancel order", e);
+ }
+
+ Map<String, Object> result = new HashMap<>();
+ result.put("code", "S");
+ result.put("orderId", orderId != null ? orderId : "unknown");
+ result.put("userId", userId);
+ return result;
+ }
+
+ public void destroy() {
+ if (failureRandom != null) {
+ failureRandom.remove();
+ }
+ }
+
+ private BigDecimal toBigDecimal(Object amountObj) {
+ return amountObj instanceof BigDecimal ? (BigDecimal) amountObj : new
BigDecimal(amountObj.toString());
+ }
+
+ private void simulateDelay() {
+ if (simulatedDelayMs > 0) {
+ try {
+
Thread.sleep(ThreadLocalRandom.current().nextInt(simulatedDelayMs));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
+ private boolean shouldFail() {
+ return failInjectionEnabled && nextFailurePercent() <
rollbackPercentage;
+ }
+
+ private void simulateTimeout(String operation) {
+ try {
+ Thread.sleep(timeoutMs);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ throw new RuntimeException("Simulated " + operation + " timeout");
+ }
+
+ private int nextFailurePercent() {
+ return FailureRandomProvider.nextPercent(failureRandom);
+ }
+}
diff --git
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/OrderSagaService.java
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/OrderSagaService.java
index 4a08b958fe..95c9d091f9 100644
---
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/OrderSagaService.java
+++
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/OrderSagaService.java
@@ -21,6 +21,7 @@ import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.Map;
+import java.util.Random;
import java.util.UUID;
import java.util.concurrent.ThreadLocalRandom;
@@ -34,10 +35,24 @@ public class OrderSagaService {
private final int rollbackPercentage;
private final int simulatedDelayMs;
-
- public OrderSagaService(int rollbackPercentage, int simulatedDelayMs) {
+ private final boolean failInjectionEnabled;
+ private final ThreadLocal<Random> failureRandom;
+ private final boolean timeoutInjectionEnabled;
+ private final int timeoutMs;
+
+ public OrderSagaService(
+ int rollbackPercentage,
+ int simulatedDelayMs,
+ boolean failInjectionEnabled,
+ ThreadLocal<Random> failureRandom,
+ boolean timeoutInjectionEnabled,
+ int timeoutMs) {
this.rollbackPercentage = rollbackPercentage;
this.simulatedDelayMs = simulatedDelayMs;
+ this.failInjectionEnabled = failInjectionEnabled;
+ this.failureRandom = failureRandom;
+ this.timeoutInjectionEnabled = timeoutInjectionEnabled;
+ this.timeoutMs = timeoutMs;
}
/**
@@ -58,7 +73,11 @@ public class OrderSagaService {
// Simulate processing time
simulateDelay();
- // Simulate random failure based on rollback percentage
+ if (timeoutInjectionEnabled) {
+ simulateTimeout("order creation");
+ }
+
+ // Simulate failure injection on the selected step.
if (shouldFail()) {
LOGGER.debug("Order creation failed (simulated): userId={}",
userId);
throw new RuntimeException("Simulated order creation failure");
@@ -95,6 +114,12 @@ public class OrderSagaService {
return result;
}
+ public void destroy() {
+ if (failureRandom != null) {
+ failureRandom.remove();
+ }
+ }
+
private void simulateDelay() {
if (simulatedDelayMs > 0) {
try {
@@ -106,6 +131,23 @@ public class OrderSagaService {
}
private boolean shouldFail() {
- return ThreadLocalRandom.current().nextInt(100) < rollbackPercentage;
+ if (!failInjectionEnabled) {
+ return false;
+ }
+ return nextFailurePercent() < rollbackPercentage;
+ }
+
+ private void simulateTimeout(String operation) {
+ LOGGER.debug("Simulating {} timeout: {} ms", operation, timeoutMs);
+ try {
+ Thread.sleep(timeoutMs);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ throw new RuntimeException("Simulated " + operation + " timeout");
+ }
+
+ private int nextFailurePercent() {
+ return FailureRandomProvider.nextPercent(failureRandom);
}
}
diff --git
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/PaymentDbSagaService.java
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/PaymentDbSagaService.java
new file mode 100644
index 0000000000..8428eae01f
--- /dev/null
+++
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/PaymentDbSagaService.java
@@ -0,0 +1,179 @@
+/*
+ * 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.seata.benchmark.saga;
+
+import javax.sql.DataSource;
+import java.math.BigDecimal;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Random;
+import java.util.concurrent.ThreadLocalRandom;
+
+/**
+ * Database-backed payment service for Saga benchmark.
+ * @author zihenzzz
+ */
+public class PaymentDbSagaService {
+
+ private final DataSource dataSource;
+ private final int rollbackPercentage;
+ private final int simulatedDelayMs;
+ private final boolean failInjectionEnabled;
+ private final ThreadLocal<Random> failureRandom;
+ private final boolean timeoutInjectionEnabled;
+ private final int timeoutMs;
+
+ public PaymentDbSagaService(
+ DataSource dataSource,
+ int rollbackPercentage,
+ int simulatedDelayMs,
+ boolean failInjectionEnabled,
+ ThreadLocal<Random> failureRandom,
+ boolean timeoutInjectionEnabled,
+ int timeoutMs) {
+ this.dataSource = dataSource;
+ this.rollbackPercentage = rollbackPercentage;
+ this.simulatedDelayMs = simulatedDelayMs;
+ this.failInjectionEnabled = failInjectionEnabled;
+ this.failureRandom = failureRandom;
+ this.timeoutInjectionEnabled = timeoutInjectionEnabled;
+ this.timeoutMs = timeoutMs;
+ }
+
+ public Map<String, Object> debitPayment(Map<String, Object> params) {
+ String accountId = (String) params.get("accountId");
+ BigDecimal amount = toBigDecimal(params.get("amount"));
+
+ simulateDelay();
+ if (timeoutInjectionEnabled) {
+ simulateTimeout("payment debit");
+ }
+ if (shouldFail()) {
+ throw new RuntimeException("Simulated payment debit failure");
+ }
+
+ try (Connection conn = dataSource.getConnection()) {
+ boolean originalAutoCommit = conn.getAutoCommit();
+ conn.setAutoCommit(false);
+ try {
+ try (PreparedStatement pstmt =
+ conn.prepareStatement("UPDATE benchmark_account SET
balance = balance - ? "
+ + "WHERE account_id = ? AND balance >= ?")) {
+ pstmt.setBigDecimal(1, amount);
+ pstmt.setString(2, accountId);
+ pstmt.setBigDecimal(3, amount);
+ if (pstmt.executeUpdate() == 0) {
+ throw new RuntimeException("Insufficient balance for
account " + accountId);
+ }
+ }
+ conn.commit();
+ } catch (Exception e) {
+ rollbackQuietly(conn, e);
+ throw e;
+ } finally {
+ restoreAutoCommit(conn, originalAutoCommit);
+ }
+ } catch (SQLException e) {
+ throw new RuntimeException("Failed to debit payment", e);
+ }
+
+ Map<String, Object> result = new HashMap<>();
+ result.put("code", "S");
+ result.put("accountId", accountId);
+ result.put("amount", amount.toPlainString());
+ return result;
+ }
+
+ public Map<String, Object> refundPayment(Map<String, Object> params) {
+ String accountId = (String) params.get("accountId");
+ BigDecimal amount = toBigDecimal(params.get("amount"));
+
+ simulateDelay();
+
+ try (Connection conn = dataSource.getConnection();
+ PreparedStatement pstmt = conn.prepareStatement(
+ "UPDATE benchmark_account SET balance = balance + ?
WHERE account_id = ?")) {
+ pstmt.setBigDecimal(1, amount);
+ pstmt.setString(2, accountId);
+ pstmt.executeUpdate();
+ } catch (SQLException e) {
+ throw new RuntimeException("Failed to refund payment", e);
+ }
+
+ Map<String, Object> result = new HashMap<>();
+ result.put("code", "S");
+ result.put("accountId", accountId);
+ result.put("amount", amount.toPlainString());
+ return result;
+ }
+
+ public void destroy() {
+ if (failureRandom != null) {
+ failureRandom.remove();
+ }
+ }
+
+ private void rollbackQuietly(Connection conn, Exception original) {
+ try {
+ conn.rollback();
+ } catch (SQLException rollbackException) {
+ original.addSuppressed(rollbackException);
+ }
+ }
+
+ private void restoreAutoCommit(Connection conn, boolean
originalAutoCommit) {
+ try {
+ conn.setAutoCommit(originalAutoCommit);
+ } catch (SQLException e) {
+ throw new RuntimeException("Failed to restore auto-commit for
payment connection", e);
+ }
+ }
+
+ private BigDecimal toBigDecimal(Object amountObj) {
+ return amountObj instanceof BigDecimal ? (BigDecimal) amountObj : new
BigDecimal(amountObj.toString());
+ }
+
+ private void simulateDelay() {
+ if (simulatedDelayMs > 0) {
+ try {
+
Thread.sleep(ThreadLocalRandom.current().nextInt(simulatedDelayMs));
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ }
+ }
+
+ private boolean shouldFail() {
+ return failInjectionEnabled && nextFailurePercent() <
rollbackPercentage;
+ }
+
+ private void simulateTimeout(String operation) {
+ try {
+ Thread.sleep(timeoutMs);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ throw new RuntimeException("Simulated " + operation + " timeout");
+ }
+
+ private int nextFailurePercent() {
+ return FailureRandomProvider.nextPercent(failureRandom);
+ }
+}
diff --git
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/PaymentSagaService.java
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/PaymentSagaService.java
index fe781925df..4abefab69f 100644
---
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/PaymentSagaService.java
+++
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/PaymentSagaService.java
@@ -22,6 +22,7 @@ import org.slf4j.LoggerFactory;
import java.math.BigDecimal;
import java.util.HashMap;
import java.util.Map;
+import java.util.Random;
import java.util.concurrent.ThreadLocalRandom;
/**
@@ -34,10 +35,24 @@ public class PaymentSagaService {
private final int rollbackPercentage;
private final int simulatedDelayMs;
-
- public PaymentSagaService(int rollbackPercentage, int simulatedDelayMs) {
+ private final boolean failInjectionEnabled;
+ private final ThreadLocal<Random> failureRandom;
+ private final boolean timeoutInjectionEnabled;
+ private final int timeoutMs;
+
+ public PaymentSagaService(
+ int rollbackPercentage,
+ int simulatedDelayMs,
+ boolean failInjectionEnabled,
+ ThreadLocal<Random> failureRandom,
+ boolean timeoutInjectionEnabled,
+ int timeoutMs) {
this.rollbackPercentage = rollbackPercentage;
this.simulatedDelayMs = simulatedDelayMs;
+ this.failInjectionEnabled = failInjectionEnabled;
+ this.failureRandom = failureRandom;
+ this.timeoutInjectionEnabled = timeoutInjectionEnabled;
+ this.timeoutMs = timeoutMs;
}
/**
@@ -57,7 +72,11 @@ public class PaymentSagaService {
// Simulate processing time
simulateDelay();
- // Simulate random failure based on rollback percentage
+ if (timeoutInjectionEnabled) {
+ simulateTimeout("payment debit");
+ }
+
+ // Simulate failure injection on the selected step.
if (shouldFail()) {
LOGGER.debug("Payment debit failed (simulated): accountId={}",
accountId);
throw new RuntimeException("Simulated payment debit failure");
@@ -96,6 +115,12 @@ public class PaymentSagaService {
return result;
}
+ public void destroy() {
+ if (failureRandom != null) {
+ failureRandom.remove();
+ }
+ }
+
private void simulateDelay() {
if (simulatedDelayMs > 0) {
try {
@@ -107,6 +132,23 @@ public class PaymentSagaService {
}
private boolean shouldFail() {
- return ThreadLocalRandom.current().nextInt(100) < rollbackPercentage;
+ if (!failInjectionEnabled) {
+ return false;
+ }
+ return nextFailurePercent() < rollbackPercentage;
+ }
+
+ private void simulateTimeout(String operation) {
+ LOGGER.debug("Simulating {} timeout: {} ms", operation, timeoutMs);
+ try {
+ Thread.sleep(timeoutMs);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ throw new RuntimeException("Simulated " + operation + " timeout");
+ }
+
+ private int nextFailurePercent() {
+ return FailureRandomProvider.nextPercent(failureRandom);
}
}
diff --git
a/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/SagaDbEnvironment.java
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/SagaDbEnvironment.java
new file mode 100644
index 0000000000..1b3435eba6
--- /dev/null
+++
b/test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/SagaDbEnvironment.java
@@ -0,0 +1,171 @@
+/*
+ * 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.seata.benchmark.saga;
+
+import com.zaxxer.hikari.HikariConfig;
+import com.zaxxer.hikari.HikariDataSource;
+import org.apache.seata.benchmark.config.BenchmarkConfig;
+import org.apache.seata.benchmark.constant.BenchmarkConstants;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.MySQLContainer;
+import org.testcontainers.utility.DockerImageName;
+
+import javax.sql.DataSource;
+import java.math.BigDecimal;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+/**
+ * Database-backed environment for Saga benchmark business actions.
+ * @author zihenzzz
+ */
+public class SagaDbEnvironment {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(SagaDbEnvironment.class);
+
+ private final BenchmarkConfig config;
+ private MySQLContainer<?> mysqlContainer;
+ private HikariDataSource dataSource;
+
+ public SagaDbEnvironment(BenchmarkConfig config) {
+ this.config = config;
+ }
+
+ public void init() {
+ try {
+ startMySQLContainer();
+ createDataSource();
+ initDatabase();
+ LOGGER.info(
+ "Saga DB workload initialized with {} accounts and {}
products",
+ BenchmarkConstants.ACCOUNT_COUNT,
+ BenchmarkConstants.SAGA_PRODUCT_COUNT);
+ } catch (Exception e) {
+ destroy();
+ throw e instanceof RuntimeException
+ ? (RuntimeException) e
+ : new RuntimeException("Failed to initialize Saga DB
workload", e);
+ }
+ }
+
+ public DataSource getDataSource() {
+ return dataSource;
+ }
+
+ public void destroy() {
+ if (dataSource != null && !dataSource.isClosed()) {
+ dataSource.close();
+ LOGGER.info("Saga DB DataSource closed");
+ }
+ if (mysqlContainer != null && mysqlContainer.isRunning()) {
+ mysqlContainer.stop();
+ LOGGER.info("Saga DB MySQL container stopped");
+ }
+ }
+
+ private void startMySQLContainer() {
+ LOGGER.info("Starting MySQL container for Saga DB workload...");
+ mysqlContainer = new
MySQLContainer<>(DockerImageName.parse("mysql:8.0"))
+ .withDatabaseName("benchmark_saga")
+ .withUsername("test")
+ .withPassword("test")
+ .withCommand("--character-set-server=utf8mb4",
"--collation-server=utf8mb4_unicode_ci");
+ mysqlContainer.start();
+ LOGGER.info("Saga DB MySQL container started: {}",
mysqlContainer.getJdbcUrl());
+ }
+
+ private void createDataSource() {
+ HikariConfig hikariConfig = new HikariConfig();
+ hikariConfig.setJdbcUrl(mysqlContainer.getJdbcUrl());
+ hikariConfig.setUsername(mysqlContainer.getUsername());
+ hikariConfig.setPassword(mysqlContainer.getPassword());
+ hikariConfig.setDriverClassName("com.mysql.cj.jdbc.Driver");
+ hikariConfig.setMaximumPoolSize(config.getThreads() * 2);
+ hikariConfig.setMinimumIdle(Math.max(1, config.getThreads()));
+ hikariConfig.setConnectionTimeout(30000);
+ hikariConfig.setIdleTimeout(600000);
+ hikariConfig.setMaxLifetime(1800000);
+ dataSource = new HikariDataSource(hikariConfig);
+ }
+
+ private void initDatabase() throws SQLException {
+ try (Connection conn = dataSource.getConnection();
+ Statement stmt = conn.createStatement()) {
+ stmt.execute("CREATE TABLE IF NOT EXISTS benchmark_inventory ("
+ + "product_id VARCHAR(64) PRIMARY KEY, "
+ + "available_qty INT NOT NULL, "
+ + "reserved_qty INT NOT NULL DEFAULT 0, "
+ + "updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON
UPDATE CURRENT_TIMESTAMP"
+ + ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4");
+ stmt.execute("CREATE TABLE IF NOT EXISTS benchmark_account ("
+ + "account_id VARCHAR(64) PRIMARY KEY, "
+ + "balance DECIMAL(18,2) NOT NULL, "
+ + "updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON
UPDATE CURRENT_TIMESTAMP"
+ + ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4");
+ stmt.execute("CREATE TABLE IF NOT EXISTS benchmark_order ("
+ + "order_id VARCHAR(64) PRIMARY KEY, "
+ + "user_id VARCHAR(64) NOT NULL, "
+ + "product_id VARCHAR(64) NOT NULL, "
+ + "quantity INT NOT NULL, "
+ + "amount DECIMAL(18,2) NOT NULL, "
+ + "status VARCHAR(32) NOT NULL, "
+ + "created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, "
+ + "updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON
UPDATE CURRENT_TIMESTAMP"
+ + ") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4");
+
+ stmt.execute("TRUNCATE TABLE benchmark_order");
+ stmt.execute("TRUNCATE TABLE benchmark_inventory");
+ stmt.execute("TRUNCATE TABLE benchmark_account");
+
+ seedInventory(conn);
+ seedAccounts(conn);
+ }
+ }
+
+ private void seedInventory(Connection conn) throws SQLException {
+ try (PreparedStatement pstmt = conn.prepareStatement(
+ "INSERT INTO benchmark_inventory (product_id, available_qty,
reserved_qty) VALUES (?, ?, 0)")) {
+ for (int i = 0; i < BenchmarkConstants.SAGA_PRODUCT_COUNT; i++) {
+ pstmt.setString(1, "product-" + i);
+ pstmt.setInt(2, BenchmarkConstants.SAGA_INITIAL_INVENTORY);
+ pstmt.addBatch();
+ if ((i + 1) % 100 == 0) {
+ pstmt.executeBatch();
+ }
+ }
+ pstmt.executeBatch();
+ }
+ }
+
+ private void seedAccounts(Connection conn) throws SQLException {
+ try (PreparedStatement pstmt =
+ conn.prepareStatement("INSERT INTO benchmark_account
(account_id, balance) VALUES (?, ?)")) {
+ for (int i = 0; i < BenchmarkConstants.ACCOUNT_COUNT; i++) {
+ pstmt.setString(1, "account-" + i);
+ pstmt.setBigDecimal(2,
BigDecimal.valueOf(BenchmarkConstants.SAGA_INITIAL_BALANCE));
+ pstmt.addBatch();
+ if ((i + 1) % 100 == 0) {
+ pstmt.executeBatch();
+ }
+ }
+ pstmt.executeBatch();
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
