From 792a21fa90f9d4aecfbca823fe720b83de423a62 Mon Sep 17 00:00:00 2001 From: monan <651932351@qq.com> Date: Thu, 27 Aug 2026 16:54:35 +0800 Subject: [PATCH 1/3] Speed up integration tests and fix amopAsyncSubTest silent timeout amopAsyncSubTest: subscribe once and wait for the subscription to propagate before broadcasting, then assert on a CountDownLatch. The old test re-subscribed the same topic in a loop and swallowed the resulting TimeoutException, so it always passed while wasting ~10s per run; the first broadcasts were also dropped silently because they were sent before the subscription reached the node. PrecompiledTest: reduce the CRUD stress loops from 100 to 20 iterations (300 -> 60 transactions), poll async receipts every 100ms with a 60s deadline, and assert the received receipt count instead of hanging forever on failure. AssembleTransactionProcessorTest / AssembleTransactionWithRemoteSignProcessorTest: replace the fixed 1s end-of-test sleeps with deterministic CompletableFuture.get() waits. AmopTest: shorten the subscription-propagation waits from 2s to 1s. Measured locally against a single-node air chain: total integrationTest time drops from 217s to 152s (-30%), all 57 tests pass. --- .../fisco/bcos/sdk/v3/test/amop/AmopTest.java | 91 +++++++++---------- .../v3/test/precompiled/PrecompiledTest.java | 15 +-- .../AssembleTransactionProcessorTest.java | 6 +- ...ransactionWithRemoteSignProcessorTest.java | 6 +- 4 files changed, 58 insertions(+), 60 deletions(-) diff --git a/src/integration-test/java/org/fisco/bcos/sdk/v3/test/amop/AmopTest.java b/src/integration-test/java/org/fisco/bcos/sdk/v3/test/amop/AmopTest.java index 40de95764..b7d32b056 100644 --- a/src/integration-test/java/org/fisco/bcos/sdk/v3/test/amop/AmopTest.java +++ b/src/integration-test/java/org/fisco/bcos/sdk/v3/test/amop/AmopTest.java @@ -3,10 +3,12 @@ import java.util.HashSet; import java.util.Set; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import org.fisco.bcos.sdk.jni.common.JniException; import org.fisco.bcos.sdk.v3.amop.Amop; @@ -46,59 +48,48 @@ public void amopAsyncSubTest() throws ConfigException, JniException, Interrupted amopBroadCast.start(); subAmop.start(); - ThreadPoolService threadPoolService = new ThreadPoolService("amop", 1000); + final int pubCount = 5; + CountDownLatch receiveLatch = new CountDownLatch(pubCount); + AtomicReference recvError = new AtomicReference<>(); - threadPoolService - .getThreadPool() - .execute( - () -> { - int count = 5; - while (count-- > 0) { - System.out.println( - " ====== AMOP broadcast, topic: " - + topic - + " ,msg: " - + message); - amopBroadCast.broadcastAmopMsg(topic, message.getBytes()); - try { - Thread.sleep(1000); - } catch (InterruptedException e) { - e.printStackTrace(); - } - } - }); + // subscribe once; the callback stays registered for the whole test + subAmop.subscribeTopic( + topic, + (endpoint, seq, data) -> { + System.out.println(" ==> receive message from client"); + System.out.println(" \t==> endpoint: " + endpoint); + System.out.println(" \t==> seq: " + seq); + System.out.println(" \t==> data: " + new String(data)); + if (!message.equals(new String(data))) { + recvError.compareAndSet( + null, "unexpected message: " + new String(data)); + } + subAmop.sendResponse(endpoint, seq, data); + receiveLatch.countDown(); + }); - threadPoolService - .getThreadPool() - .execute( - () -> { - int count = 5; - while (count-- > 0) { - CompletableFuture future = new CompletableFuture<>(); - subAmop.subscribeTopic( - topic, - (endpoint, seq, data) -> { - System.out.println(" ==> receive message from client"); - System.out.println(" \t==> endpoint: " + endpoint); - System.out.println(" \t==> seq: " + seq); - System.out.println(" \t==> data: " + new String(data)); - Assert.assertEquals(new String(data), message); - subAmop.sendResponse(endpoint, seq, data); - future.complete(false); - }); - try { - future.get(10, TimeUnit.SECONDS); - } catch (InterruptedException | ExecutionException | TimeoutException e) { - e.printStackTrace(); - } - } - }); - Thread.sleep(10000); - threadPoolService.stop(); + // the subscription is pushed to the node asynchronously; broadcasts sent + // before it takes effect are dropped silently, so wait for it first + Thread.sleep(3000); + + for (int i = 0; i < pubCount; i++) { + System.out.println(" ====== AMOP broadcast, topic: " + topic + " ,msg: " + message); + amopBroadCast.broadcastAmopMsg(topic, message.getBytes()); + Thread.sleep(1000); + } + + // fail the test if not all broadcasts are received, instead of swallowing + // a TimeoutException + boolean allReceived = receiveLatch.await(15, TimeUnit.SECONDS); amopBroadCast.stop(); subAmop.stop(); amopBroadCast.destroy(); subAmop.destroy(); + Assert.assertNull(recvError.get(), recvError.get()); + Assert.assertTrue( + "only received " + (pubCount - receiveLatch.getCount()) + "/" + pubCount + + " broadcast messages", + allReceived); } @Test @@ -142,7 +133,7 @@ public void amopSubAsyncTest() endpoint, seq, message2.getBytes()); })); - Thread.sleep(2000); + Thread.sleep(1000); AtomicInteger countResponse = new AtomicInteger(pubTime); CompletableFuture future = new CompletableFuture<>(); @@ -199,7 +190,7 @@ public void amopSubTest() subAmop.sendResponse(endpoint, seq, message2.getBytes()); }); - Thread.sleep(2000); + Thread.sleep(1000); AtomicInteger countResponse = new AtomicInteger(pubTime); CompletableFuture future = new CompletableFuture<>(); @@ -258,7 +249,7 @@ public void amopUnsubTest() subAmop.getSubTopics(); subAmop.unsubscribeTopic(topic); - Thread.sleep(2000); + Thread.sleep(1000); AtomicInteger countResponse = new AtomicInteger(pubTime); CompletableFuture future = new CompletableFuture<>(); diff --git a/src/integration-test/java/org/fisco/bcos/sdk/v3/test/precompiled/PrecompiledTest.java b/src/integration-test/java/org/fisco/bcos/sdk/v3/test/precompiled/PrecompiledTest.java index 2eaf48a4c..2145b14ed 100644 --- a/src/integration-test/java/org/fisco/bcos/sdk/v3/test/precompiled/PrecompiledTest.java +++ b/src/integration-test/java/org/fisco/bcos/sdk/v3/test/precompiled/PrecompiledTest.java @@ -301,7 +301,7 @@ public void test51SyncCRUDService() throws ConfigException, ContractException { client.getTotalTransactionCount() .getTotalTransactionCount() .getTransactionCount()); - for (int i = 0; i < 100; i++) { + for (int i = 0; i < 20; i++) { Integer index = i; threadPool.execute( () -> { @@ -334,7 +334,7 @@ public void test51SyncCRUDService() throws ConfigException, ContractException { .getTotalTransactionCount() .getTransactionCount()); System.out.println("orgTxCount: " + orgTxCount + ", currentTxCount:" + currentTxCount); - Assert.assertTrue(currentTxCount.compareTo(orgTxCount.add(BigInteger.valueOf(300))) >= 0); + Assert.assertTrue(currentTxCount.compareTo(orgTxCount.add(BigInteger.valueOf(60))) >= 0); client.stop(); client.destroy(); } @@ -376,7 +376,7 @@ public void test52AsyncCRUDService() client.getTotalTransactionCount() .getTotalTransactionCount() .getTransactionCount()); - for (int i = 0; i < 100; i++) { + for (int i = 0; i < 20; i++) { int index = i; threadPool.execute( () -> { @@ -406,8 +406,10 @@ public void test52AsyncCRUDService() } }); } - while (this.receiptCount.get() != 300) { - Thread.sleep(1000); + // wait for all async callbacks, but fail instead of hanging forever + long deadline = System.currentTimeMillis() + 60000; + while (this.receiptCount.get() != 60 && System.currentTimeMillis() < deadline) { + Thread.sleep(100); } ThreadPoolService.stopThreadPool(threadPool); BigInteger currentTxCount = @@ -416,7 +418,8 @@ public void test52AsyncCRUDService() .getTotalTransactionCount() .getTransactionCount()); System.out.println("orgTxCount: " + orgTxCount + ", currentTxCount:" + currentTxCount); - Assert.assertTrue(currentTxCount.compareTo(orgTxCount.add(BigInteger.valueOf(300))) >= 0); + Assert.assertEquals(60, this.receiptCount.get()); + Assert.assertTrue(currentTxCount.compareTo(orgTxCount.add(BigInteger.valueOf(60))) >= 0); client.stop(); client.destroy(); } diff --git a/src/integration-test/java/org/fisco/bcos/sdk/v3/test/transaction/manager/AssembleTransactionProcessorTest.java b/src/integration-test/java/org/fisco/bcos/sdk/v3/test/transaction/manager/AssembleTransactionProcessorTest.java index a986e47c2..f886c21fe 100644 --- a/src/integration-test/java/org/fisco/bcos/sdk/v3/test/transaction/manager/AssembleTransactionProcessorTest.java +++ b/src/integration-test/java/org/fisco/bcos/sdk/v3/test/transaction/manager/AssembleTransactionProcessorTest.java @@ -17,6 +17,7 @@ import java.math.BigInteger; import java.util.*; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; import org.apache.commons.collections4.ListUtils; import org.apache.commons.lang3.StringUtils; @@ -190,8 +191,9 @@ public void test11HelloWorldAsync() throws Exception { }); System.out.println("--- finish deploy with CompletableFuture ---"); - // wait for the async thread - Thread.sleep(1000); + // wait for the async deploy deterministically instead of a fixed sleep + TransactionReceipt receipt = future.get(10, TimeUnit.SECONDS); + Assert.assertEquals(0, receipt.getStatus()); } @Test diff --git a/src/integration-test/java/org/fisco/bcos/sdk/v3/test/transaction/manager/AssembleTransactionWithRemoteSignProcessorTest.java b/src/integration-test/java/org/fisco/bcos/sdk/v3/test/transaction/manager/AssembleTransactionWithRemoteSignProcessorTest.java index ea2f57294..e124e16ec 100644 --- a/src/integration-test/java/org/fisco/bcos/sdk/v3/test/transaction/manager/AssembleTransactionWithRemoteSignProcessorTest.java +++ b/src/integration-test/java/org/fisco/bcos/sdk/v3/test/transaction/manager/AssembleTransactionWithRemoteSignProcessorTest.java @@ -18,6 +18,7 @@ import java.util.Collections; import java.util.List; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; import org.apache.commons.lang3.StringUtils; import org.fisco.bcos.sdk.jni.utilities.tx.TransactionBuilderJniObj; @@ -212,7 +213,8 @@ public void test2HelloWorldAsync() throws Exception { return null; }); - // wait for the async thread - Thread.sleep(1000); + // wait for the async operations deterministically instead of a fixed sleep + Assert.assertEquals(0, future.get(10, TimeUnit.SECONDS).getStatus()); + Assert.assertEquals(0, future2.get(10, TimeUnit.SECONDS).getStatus()); } } From 97905d5600d31a191be80c01d11a626a6ba6916e Mon Sep 17 00:00:00 2001 From: monan <651932351@qq.com> Date: Thu, 27 Aug 2026 17:43:15 +0800 Subject: [PATCH 2/3] Count only successful responses in test52AsyncCRUDService FakeTransactionCallback counted every onResponse as a sealed tx, so error responses (e.g. client-side timeouts under load) made the receipt-count wait pass while the on-chain tx count assertion failed with no diagnostics. Count only successful receipts and log the error code/message otherwise. --- .../sdk/v3/test/precompiled/PrecompiledTest.java | 13 +++++++++++-- 1 file changed, 11 insertions(+), 2 deletions(-) diff --git a/src/integration-test/java/org/fisco/bcos/sdk/v3/test/precompiled/PrecompiledTest.java b/src/integration-test/java/org/fisco/bcos/sdk/v3/test/precompiled/PrecompiledTest.java index 2145b14ed..38df87987 100644 --- a/src/integration-test/java/org/fisco/bcos/sdk/v3/test/precompiled/PrecompiledTest.java +++ b/src/integration-test/java/org/fisco/bcos/sdk/v3/test/precompiled/PrecompiledTest.java @@ -342,11 +342,20 @@ public void test51SyncCRUDService() throws ConfigException, ContractException { class FakeTransactionCallback implements PrecompiledCallback { public TransactionReceipt receipt; - // wait until get the transactionReceipt + // wait until get the transactionReceipt; only successful responses count, + // otherwise an error response would be silently treated as a sealed tx @Override public void onResponse(RetCode retCode) { this.receipt = retCode.getTransactionReceipt(); - PrecompiledTest.this.receiptCount.addAndGet(1); + if (retCode.getCode() == 0 && this.receipt != null && this.receipt.isStatusOK()) { + PrecompiledTest.this.receiptCount.addAndGet(1); + } else { + System.out.println( + "async crud failed, code: " + + retCode.getCode() + + ", message: " + + retCode.getMessage()); + } } } From 9bee990ec6eeb0c532d357e873e83b2534aa7b81 Mon Sep 17 00:00:00 2001 From: monan <651932351@qq.com> Date: Thu, 27 Aug 2026 18:31:16 +0800 Subject: [PATCH 3/3] Make test52AsyncCRUDService deterministic The old test fired asyncInsert + asyncUpdate + asyncRemove for the same key simultaneously, so update/remove routinely landed before the insert and failed with -51507/-51508; the callback also treated every response as success, hiding those failures completely. Chain the three calls per key through the callbacks (the next call is submitted to the test thread pool, since resolving the table address is a blocking call that must not run on the sdk callback thread), count receipts by receipt status (a successful CRUD retCode carries the affected row count, e.g. 1, not 0), and keep the 60s deadline so a lost callback fails the test instead of hanging. --- .../v3/test/precompiled/PrecompiledTest.java | 72 ++++++++++++++----- 1 file changed, 55 insertions(+), 17 deletions(-) diff --git a/src/integration-test/java/org/fisco/bcos/sdk/v3/test/precompiled/PrecompiledTest.java b/src/integration-test/java/org/fisco/bcos/sdk/v3/test/precompiled/PrecompiledTest.java index 38df87987..140e59f01 100644 --- a/src/integration-test/java/org/fisco/bcos/sdk/v3/test/precompiled/PrecompiledTest.java +++ b/src/integration-test/java/org/fisco/bcos/sdk/v3/test/precompiled/PrecompiledTest.java @@ -341,20 +341,35 @@ public void test51SyncCRUDService() throws ConfigException, ContractException { class FakeTransactionCallback implements PrecompiledCallback { public TransactionReceipt receipt; + private final ExecutorService executor; + private final Runnable onSuccess; - // wait until get the transactionReceipt; only successful responses count, - // otherwise an error response would be silently treated as a sealed tx + FakeTransactionCallback(ExecutorService executor, Runnable onSuccess) { + this.executor = executor; + this.onSuccess = onSuccess; + } + + // wait until get the transactionReceipt; count by the receipt status only: + // for CRUD precompiled calls a successful retCode carries the affected row + // count (e.g. 1 for a successful insert), not 0 @Override public void onResponse(RetCode retCode) { this.receipt = retCode.getTransactionReceipt(); - if (retCode.getCode() == 0 && this.receipt != null && this.receipt.isStatusOK()) { + if (this.receipt != null && this.receipt.isStatusOK()) { PrecompiledTest.this.receiptCount.addAndGet(1); + if (onSuccess != null) { + // the next CRUD call resolves the table address with a blocking + // call, it must not run on the sdk callback thread + executor.execute(onSuccess); + } } else { System.out.println( "async crud failed, code: " + retCode.getCode() + ", message: " - + retCode.getMessage()); + + retCode.getMessage() + + ", receipt status: " + + (this.receipt == null ? "null" : this.receipt.getStatus())); } } } @@ -392,22 +407,45 @@ public void test52AsyncCRUDService() try { LinkedHashMap value = new LinkedHashMap<>(); value.put("field", "field" + index); - // insert - FakeTransactionCallback callback = new FakeTransactionCallback(); + // chain insert -> update -> remove per key through the callbacks, + // firing them together races and fails with "Key not exist" crudService.asyncInsert( tableName, new Entry(valueFiled, "key" + index, value), - callback); - // update - value.clear(); - value.put("field", "field" + index + 100); - UpdateFields updateFields = new UpdateFields(value); - FakeTransactionCallback callback2 = new FakeTransactionCallback(); - crudService.asyncUpdate( - tableName, "key" + index, updateFields, callback2); - // remove - FakeTransactionCallback callback3 = new FakeTransactionCallback(); - crudService.asyncRemove(tableName, "key" + index, callback3); + new FakeTransactionCallback( + threadPool, + () -> { + try { + LinkedHashMap newValue = + new LinkedHashMap<>(); + newValue.put("field", "field" + index + 100); + crudService.asyncUpdate( + tableName, + "key" + index, + new UpdateFields(newValue), + new FakeTransactionCallback( + threadPool, + () -> { + try { + crudService.asyncRemove( + tableName, + "key" + index, + new FakeTransactionCallback( + threadPool, + null)); + } catch (ContractException + e) { + System.out.println( + "asyncRemove failed: " + + e.getMessage()); + } + })); + } catch (ContractException e) { + System.out.println( + "asyncUpdate failed: " + + e.getMessage()); + } + })); } catch (ContractException e) { System.out.println( "call crudService failed, error information: "