Skip to content

Speed up integration tests and fix amopAsyncSubTest silent timeout #963

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
morebtcg wants to merge 3 commits into
base: master
Choose a base branch
from
Open

Speed up integration tests and fix amopAsyncSubTest silent timeout #963

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
792a21f
Speed up integration tests and fix amopAsyncSubTest silent timeout
morebtcg Aug 27, 2026
97905d5
Count only successful responses in test52AsyncCRUDService
morebtcg Aug 27, 2026
9bee990
Make test52AsyncCRUDService deterministic
morebtcg Aug 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
91 changes: 41 additions & 50 deletions src/integration-test/java/org/fisco/bcos/sdk/v3/test/amop/AmopTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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<String> 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<Boolean> 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
Expand Down Expand Up @@ -142,7 +133,7 @@ public void amopSubAsyncTest()
endpoint, seq, message2.getBytes());
}));

Thread.sleep(2000);
Thread.sleep(1000);

AtomicInteger countResponse = new AtomicInteger(pubTime);
CompletableFuture<Boolean> future = new CompletableFuture<>();
Expand Down Expand Up @@ -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<Boolean> future = new CompletableFuture<>();
Expand Down Expand Up @@ -258,7 +249,7 @@ public void amopUnsubTest()
subAmop.getSubTopics();
subAmop.unsubscribeTopic(topic);

Thread.sleep(2000);
Thread.sleep(1000);

AtomicInteger countResponse = new AtomicInteger(pubTime);
CompletableFuture<Boolean> future = new CompletableFuture<>();
Expand Down
92 changes: 71 additions & 21 deletions src/integration-test/java/org/fisco/bcos/sdk/v3/test/precompiled/PrecompiledTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(
() -> {
Expand Down Expand Up @@ -334,19 +334,43 @@ 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();
}

class FakeTransactionCallback implements PrecompiledCallback {
public TransactionReceipt receipt;
private final ExecutorService executor;
private final Runnable onSuccess;

// wait until get the transactionReceipt
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();
PrecompiledTest.this.receiptCount.addAndGet(1);
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()
+ ", receipt status: "
+ (this.receipt == null ? "null" : this.receipt.getStatus()));
}
}
}

Expand Down Expand Up @@ -376,38 +400,63 @@ 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(
() -> {
try {
LinkedHashMap<String, String> 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<String, String> 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: "
+ e.getMessage());
}
});
}
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 =
Expand All @@ -416,7 +465,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();
}
Expand Down
6 changes: 4 additions & 2 deletions ...java/org/fisco/bcos/sdk/v3/test/transaction/manager/AssembleTransactionProcessorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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
Expand Down
6 changes: 4 additions & 2 deletions .../bcos/sdk/v3/test/transaction/manager/AssembleTransactionWithRemoteSignProcessorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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());
}
}
Loading