Copilot commented on code in PR #8042:
URL: https://github.com/apache/incubator-seata/pull/8042#discussion_r3069106763
##########
test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/model/BenchmarkMetrics.java:
##########
@@ -80,6 +107,26 @@ public long getFailedCount() {
return failedCount.get();
}
+ public long getCommittedCount() {
+ return committedCount.get();
+ }
+
+ public long getCompensatedCount() {
+ return compensatedCount.get();
+ }
+
+ public long getExecutionFailedCount() {
+ return failedCount.get() - compensatedCount.get() -
compensationFailedCount.get() - unknownCount.get();
Review Comment:
`getExecutionFailedCount()` subtracts `compensatedCount` from `failedCount`,
but compensated transactions are recorded as successes (they don’t increment
`failedCount`). This can undercount execution failures and even return a
negative number when many transactions are compensated. Compute execution
failures from the failure-side buckets only (e.g., failed minus
compensationFailed minus unknown), or track an explicit executionFailed counter.
```suggestion
return failedCount.get() - compensationFailedCount.get() -
unknownCount.get();
```
##########
test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/PaymentDbSagaService.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 javax.sql.DataSource;
+import java.math.BigDecimal;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+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 Random failureRandom;
+ private final boolean timeoutInjectionEnabled;
+ private final int timeoutMs;
+
+ public PaymentDbSagaService(
+ DataSource dataSource,
+ int rollbackPercentage,
+ int simulatedDelayMs,
+ boolean failInjectionEnabled,
+ 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()) {
+ conn.setAutoCommit(false);
+ BigDecimal balance = queryBalance(conn, accountId);
+ if (balance.compareTo(amount) < 0) {
+ throw new RuntimeException("Insufficient balance for account "
+ accountId);
+ }
+ try (PreparedStatement pstmt = conn.prepareStatement(
+ "UPDATE benchmark_account SET balance = balance - ? WHERE
account_id = ?")) {
+ pstmt.setBigDecimal(1, amount);
+ pstmt.setString(2, accountId);
+ pstmt.executeUpdate();
+ }
+ conn.commit();
Review Comment:
This manual transaction doesn’t roll back on non-SQL exceptions (e.g.,
insufficient balance) before the connection is returned to the pool. Add a
try/catch/finally that performs `rollback()` on any throwable after
`setAutoCommit(false)` and restores connection state to avoid leaking an open
transaction into the pool.
```suggestion
boolean originalAutoCommit = conn.getAutoCommit();
Throwable pending = null;
conn.setAutoCommit(false);
try {
BigDecimal balance = queryBalance(conn, accountId);
if (balance.compareTo(amount) < 0) {
throw new RuntimeException("Insufficient balance for
account " + accountId);
}
try (PreparedStatement pstmt = conn.prepareStatement(
"UPDATE benchmark_account SET balance = balance - ?
WHERE account_id = ?")) {
pstmt.setBigDecimal(1, amount);
pstmt.setString(2, accountId);
pstmt.executeUpdate();
}
conn.commit();
} catch (Throwable t) {
pending = t;
try {
conn.rollback();
} catch (SQLException rollbackException) {
t.addSuppressed(rollbackException);
}
throw t;
} finally {
try {
conn.setAutoCommit(originalAutoCommit);
} catch (SQLException resetException) {
if (pending != null) {
pending.addSuppressed(resetException);
} else {
throw new RuntimeException("Failed to restore
auto-commit", resetException);
}
}
}
```
##########
test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/PaymentSagaService.java:
##########
@@ -107,6 +126,28 @@ private void simulateDelay() {
}
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() {
+ if (failureRandom != null) {
+ synchronized (failureRandom) {
+ return failureRandom.nextInt(100);
+ }
+ }
+ return ThreadLocalRandom.current().nextInt(100);
Review Comment:
`nextFailurePercent()` synchronizes on the shared `Random` instance. With
multiple benchmark threads this introduces lock contention and can distort
benchmark results. Prefer a per-thread RNG (e.g., a `ThreadLocal` RNG derived
from the configured seed, or a `SplittableRandom` per thread) so failure
injection remains reproducible without a global lock.
##########
test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/InventorySagaService.java:
##########
@@ -102,6 +121,28 @@ private void simulateDelay() {
}
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() {
+ if (failureRandom != null) {
+ synchronized (failureRandom) {
+ return failureRandom.nextInt(100);
+ }
+ }
+ return ThreadLocalRandom.current().nextInt(100);
Review Comment:
`nextFailurePercent()` synchronizes on the shared `Random` instance. With
multiple benchmark threads this introduces lock contention and can distort
benchmark results. Prefer a per-thread RNG (e.g., a `ThreadLocal` RNG derived
from the configured seed, or a `SplittableRandom` per thread) so failure
injection remains reproducible without a global lock.
##########
test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/InventoryDbSagaService.java:
##########
@@ -0,0 +1,178 @@
+/*
+ * 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.ResultSet;
+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 Random failureRandom;
+ private final boolean timeoutInjectionEnabled;
+ private final int timeoutMs;
+
+ public InventoryDbSagaService(
+ DataSource dataSource,
+ int rollbackPercentage,
+ int simulatedDelayMs,
+ boolean failInjectionEnabled,
+ 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()) {
+ conn.setAutoCommit(false);
+ int available = queryAvailableQuantity(conn, productId);
+ if (available < quantity) {
+ throw new RuntimeException("Insufficient inventory for product
" + productId);
+ }
+ try (PreparedStatement pstmt = conn.prepareStatement(
+ "UPDATE benchmark_inventory "
+ + "SET available_qty = available_qty - ?,
reserved_qty = reserved_qty + ? "
+ + "WHERE product_id = ?")) {
+ pstmt.setInt(1, quantity);
+ pstmt.setInt(2, quantity);
+ pstmt.setString(3, productId);
+ pstmt.executeUpdate();
+ }
+ conn.commit();
+ } 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;
+ }
+
+ private int queryAvailableQuantity(Connection conn, String productId)
throws SQLException {
+ try (PreparedStatement pstmt =
+ conn.prepareStatement("SELECT available_qty FROM
benchmark_inventory WHERE product_id = ?");
+ ResultSet rs = executeQuery(pstmt, productId)) {
+ if (!rs.next()) {
+ throw new RuntimeException("Product not found: " + productId);
+ }
+ return rs.getInt(1);
+ }
+ }
+
+ private ResultSet executeQuery(PreparedStatement pstmt, String productId)
throws SQLException {
+ pstmt.setString(1, productId);
+ return pstmt.executeQuery();
+ }
+
+ 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() {
+ if (failureRandom != null) {
+ synchronized (failureRandom) {
+ return failureRandom.nextInt(100);
+ }
+ }
+ return ThreadLocalRandom.current().nextInt(100);
+ }
Review Comment:
`nextFailurePercent()` synchronizes on the shared `Random` instance. With
multiple benchmark threads this introduces lock contention and can distort
benchmark results. Prefer a per-thread RNG (e.g., a `ThreadLocal` RNG derived
from the configured seed, or a `SplittableRandom` per thread) so failure
injection remains reproducible without a global lock.
##########
test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/InventoryDbSagaService.java:
##########
@@ -0,0 +1,178 @@
+/*
+ * 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.ResultSet;
+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 Random failureRandom;
+ private final boolean timeoutInjectionEnabled;
+ private final int timeoutMs;
+
+ public InventoryDbSagaService(
+ DataSource dataSource,
+ int rollbackPercentage,
+ int simulatedDelayMs,
+ boolean failInjectionEnabled,
+ 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()) {
+ conn.setAutoCommit(false);
+ int available = queryAvailableQuantity(conn, productId);
+ if (available < quantity) {
+ throw new RuntimeException("Insufficient inventory for product
" + productId);
+ }
+ try (PreparedStatement pstmt = conn.prepareStatement(
+ "UPDATE benchmark_inventory "
+ + "SET available_qty = available_qty - ?,
reserved_qty = reserved_qty + ? "
+ + "WHERE product_id = ?")) {
+ pstmt.setInt(1, quantity);
+ pstmt.setInt(2, quantity);
+ pstmt.setString(3, productId);
+ pstmt.executeUpdate();
Review Comment:
The inventory availability check is a separate read followed by an update
that unconditionally decrements `available_qty`. With concurrent reservations
this can drive `available_qty` negative or allow overselling. Consider locking
the row during the check (e.g., `SELECT ... FOR UPDATE`) or using a single
conditional update (`... SET available_qty = available_qty - ? ... WHERE
product_id = ? AND available_qty >= ?`) and failing if no row is updated.
```suggestion
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);
int updatedRows = pstmt.executeUpdate();
if (updatedRows == 0) {
throw new RuntimeException("Insufficient inventory for
product " + productId);
}
```
##########
test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/InventoryDbSagaService.java:
##########
@@ -0,0 +1,178 @@
+/*
+ * 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.ResultSet;
+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 Random failureRandom;
+ private final boolean timeoutInjectionEnabled;
+ private final int timeoutMs;
+
+ public InventoryDbSagaService(
+ DataSource dataSource,
+ int rollbackPercentage,
+ int simulatedDelayMs,
+ boolean failInjectionEnabled,
+ 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()) {
+ conn.setAutoCommit(false);
+ int available = queryAvailableQuantity(conn, productId);
+ if (available < quantity) {
+ throw new RuntimeException("Insufficient inventory for product
" + productId);
+ }
+ try (PreparedStatement pstmt = conn.prepareStatement(
+ "UPDATE benchmark_inventory "
+ + "SET available_qty = available_qty - ?,
reserved_qty = reserved_qty + ? "
+ + "WHERE product_id = ?")) {
+ pstmt.setInt(1, quantity);
+ pstmt.setInt(2, quantity);
+ pstmt.setString(3, productId);
+ pstmt.executeUpdate();
+ }
+ conn.commit();
Review Comment:
This manual transaction doesn’t roll back on non-SQL exceptions (e.g.,
insufficient inventory) before the connection is returned to the pool. Add a
try/catch/finally that performs `rollback()` on any throwable after
`setAutoCommit(false)` and restores connection state to avoid leaking an open
transaction into the pool.
```suggestion
boolean originalAutoCommit = conn.getAutoCommit();
boolean restoreAutoCommit = false;
try {
conn.setAutoCommit(false);
restoreAutoCommit = true;
int available = queryAvailableQuantity(conn, productId);
if (available < quantity) {
throw new RuntimeException("Insufficient inventory for
product " + productId);
}
try (PreparedStatement pstmt = conn.prepareStatement(
"UPDATE benchmark_inventory "
+ "SET available_qty = available_qty - ?,
reserved_qty = reserved_qty + ? "
+ "WHERE product_id = ?")) {
pstmt.setInt(1, quantity);
pstmt.setInt(2, quantity);
pstmt.setString(3, productId);
pstmt.executeUpdate();
}
conn.commit();
} catch (Throwable t) {
if (restoreAutoCommit) {
try {
conn.rollback();
} catch (SQLException rollbackEx) {
LOGGER.warn("Failed to rollback inventory
reservation transaction for productId={}",
productId, rollbackEx);
}
}
if (t instanceof SQLException) {
throw new RuntimeException("Failed to reserve
inventory", t);
}
throw t;
} finally {
if (restoreAutoCommit) {
try {
conn.setAutoCommit(originalAutoCommit);
} catch (SQLException resetEx) {
LOGGER.warn("Failed to restore auto-commit for
inventory reservation connection, productId={}",
productId, resetEx);
}
}
}
```
##########
test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/PaymentDbSagaService.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 javax.sql.DataSource;
+import java.math.BigDecimal;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+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 Random failureRandom;
+ private final boolean timeoutInjectionEnabled;
+ private final int timeoutMs;
+
+ public PaymentDbSagaService(
+ DataSource dataSource,
+ int rollbackPercentage,
+ int simulatedDelayMs,
+ boolean failInjectionEnabled,
+ 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()) {
+ conn.setAutoCommit(false);
+ BigDecimal balance = queryBalance(conn, accountId);
+ if (balance.compareTo(amount) < 0) {
+ throw new RuntimeException("Insufficient balance for account "
+ accountId);
+ }
+ try (PreparedStatement pstmt = conn.prepareStatement(
+ "UPDATE benchmark_account SET balance = balance - ? WHERE
account_id = ?")) {
+ pstmt.setBigDecimal(1, amount);
+ pstmt.setString(2, accountId);
+ pstmt.executeUpdate();
Review Comment:
The balance check is done as a separate read (`SELECT balance`) followed by
an `UPDATE balance = balance - ?`. Under concurrency this can overdraw accounts
(race between the check and the update). Consider a single conditional update
(e.g., update with `WHERE balance >= ?`) and treat 0 updated rows as
insufficient funds, or lock the row (e.g., `SELECT ... FOR UPDATE`) within the
transaction.
```suggestion
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);
int updatedRows = pstmt.executeUpdate();
if (updatedRows == 0) {
throw new RuntimeException("Insufficient balance for
account " + accountId);
}
```
##########
test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/OrderSagaService.java:
##########
@@ -106,6 +125,28 @@ private void simulateDelay() {
}
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() {
+ if (failureRandom != null) {
+ synchronized (failureRandom) {
+ return failureRandom.nextInt(100);
+ }
+ }
+ return ThreadLocalRandom.current().nextInt(100);
Review Comment:
`nextFailurePercent()` synchronizes on the shared `Random` instance. With
multiple benchmark threads this introduces lock contention and can distort
benchmark results. Prefer a per-thread RNG (e.g., a `ThreadLocal` RNG derived
from the configured seed, or a `SplittableRandom` per thread) so failure
injection remains reproducible without a global lock.
##########
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 Random failureRandom;
+ private final boolean timeoutInjectionEnabled;
+ private final int timeoutMs;
+
+ public OrderDbSagaService(
+ DataSource dataSource,
+ int rollbackPercentage,
+ int simulatedDelayMs,
+ boolean failInjectionEnabled,
+ 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;
+ }
+
+ 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() {
+ if (failureRandom != null) {
+ synchronized (failureRandom) {
+ return failureRandom.nextInt(100);
+ }
Review Comment:
`nextFailurePercent()` synchronizes on the shared `Random` instance. With
multiple benchmark threads this introduces lock contention and can distort
benchmark results. Prefer a per-thread RNG (e.g., a `ThreadLocal` RNG derived
from the configured seed, or a `SplittableRandom` per thread) so failure
injection remains reproducible without a global lock.
```suggestion
return failureRandom.nextInt(100);
```
##########
test-suite/seata-benchmark-cli/src/main/java/org/apache/seata/benchmark/saga/PaymentDbSagaService.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 javax.sql.DataSource;
+import java.math.BigDecimal;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+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 Random failureRandom;
+ private final boolean timeoutInjectionEnabled;
+ private final int timeoutMs;
+
+ public PaymentDbSagaService(
+ DataSource dataSource,
+ int rollbackPercentage,
+ int simulatedDelayMs,
+ boolean failInjectionEnabled,
+ 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()) {
+ conn.setAutoCommit(false);
+ BigDecimal balance = queryBalance(conn, accountId);
+ if (balance.compareTo(amount) < 0) {
+ throw new RuntimeException("Insufficient balance for account "
+ accountId);
+ }
+ try (PreparedStatement pstmt = conn.prepareStatement(
+ "UPDATE benchmark_account SET balance = balance - ? WHERE
account_id = ?")) {
+ pstmt.setBigDecimal(1, amount);
+ pstmt.setString(2, accountId);
+ pstmt.executeUpdate();
+ }
+ conn.commit();
+ } 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;
+ }
+
+ private BigDecimal queryBalance(Connection conn, String accountId) throws
SQLException {
+ try (PreparedStatement pstmt =
+ conn.prepareStatement("SELECT balance FROM
benchmark_account WHERE account_id = ?");
+ ResultSet rs = executeQuery(pstmt, accountId)) {
+ if (!rs.next()) {
+ throw new RuntimeException("Account not found: " + accountId);
+ }
+ return rs.getBigDecimal(1);
+ }
+ }
+
+ private ResultSet executeQuery(PreparedStatement pstmt, String accountId)
throws SQLException {
+ pstmt.setString(1, accountId);
+ return pstmt.executeQuery();
+ }
+
+ 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() {
+ if (failureRandom != null) {
+ synchronized (failureRandom) {
+ return failureRandom.nextInt(100);
+ }
+ }
+ return ThreadLocalRandom.current().nextInt(100);
+ }
Review Comment:
`nextFailurePercent()` synchronizes on the shared `Random` instance. With
multiple benchmark threads this introduces lock contention and can distort
benchmark results. Prefer a per-thread RNG (e.g., a `ThreadLocal` RNG derived
from the configured seed, or a `SplittableRandom` per thread) so failure
injection remains reproducible without a global lock.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
