CRZbulabula commented on code in PR #1475:
URL: https://github.com/apache/ratis/pull/1475#discussion_r3329719365
##########
ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java:
##########
@@ -127,46 +128,41 @@ static void runTestBasicAppendEntries(
final CompletableFuture<Void> killAndRestartFollower =
killAndRestartServer(
cluster.getFollowers().get(0).getId(), 0, 1000, cluster, log);
- final CompletableFuture<Void> killAndRestartLeader;
- if (killLeader) {
- log.info("killAndRestart leader " + leader.getId());
- killAndRestartLeader = killAndRestartServer(leader.getId(), 2000, 4000,
cluster, log);
- } else {
- killAndRestartLeader = CompletableFuture.completedFuture(null);
- }
-
- log.info(cluster.printServers());
+ CompletableFuture<Void> killAndRestartLeader =
CompletableFuture.completedFuture(null);
final SimpleMessage[] messages = SimpleMessage.create(numMessages);
- try (final RaftClient client = cluster.createClient()) {
- final AtomicInteger asyncReplyCount = new AtomicInteger();
- final CompletableFuture<Void> f = new CompletableFuture<>();
+ try {
+ log.info(cluster.printServers());
+
+ try (final RaftClient client = cluster.createClient()) {
+ final List<CompletableFuture<RaftClientReply>> asyncReplies = new
ArrayList<>();
- for (SimpleMessage message : messages) {
+ for (SimpleMessage message : messages) {
+ if (async) {
+ asyncReplies.add(client.async().send(message));
+ } else {
+ final RaftClientReply reply = client.io().send(message);
+ Assertions.assertTrue(reply.isSuccess());
+ }
+ }
if (async) {
- client.async().send(message).thenAcceptAsync(reply -> {
- if (!reply.isSuccess()) {
- f.completeExceptionally(
- new AssertionError("Failed with reply " + reply));
- } else if (asyncReplyCount.incrementAndGet() == messages.length) {
- f.complete(null);
- }
+ CompletableFuture.allOf(asyncReplies.toArray(new
CompletableFuture<?>[0])).join();
+ asyncReplies.forEach(f -> {
+ final RaftClientReply reply = f.join();
+ Assertions.assertTrue(reply.isSuccess(), () -> "Failed with reply
" + reply);
});
- } else {
- final RaftClientReply reply = client.io().send(message);
- Assertions.assertTrue(reply.isSuccess());
}
}
- if (async) {
- f.join();
- Assertions.assertEquals(messages.length, asyncReplyCount.get());
+ if (killLeader) {
+ log.info("killAndRestart leader " + leader.getId());
+ killAndRestartLeader = killAndRestartServer(leader.getId(), 0, 4000,
cluster, log);
}
+ Thread.sleep(cluster.getTimeoutMax().toIntExact(TimeUnit.MILLISECONDS) +
100);
+ } finally {
+ CompletableFuture.allOf(killAndRestartFollower,
killAndRestartLeader).join();
}
- Thread.sleep(cluster.getTimeoutMax().toIntExact(TimeUnit.MILLISECONDS) +
100);
log.info(cluster.printAllLogs());
- killAndRestartFollower.join();
- killAndRestartLeader.join();
Review Comment:
Done. Removed the try/finally restructuring and now join the restart futures
before `cluster.printAllLogs()`.
##########
ratis-server/src/test/java/org/apache/ratis/RaftBasicTests.java:
##########
@@ -127,46 +128,41 @@ static void runTestBasicAppendEntries(
final CompletableFuture<Void> killAndRestartFollower =
killAndRestartServer(
cluster.getFollowers().get(0).getId(), 0, 1000, cluster, log);
- final CompletableFuture<Void> killAndRestartLeader;
- if (killLeader) {
- log.info("killAndRestart leader " + leader.getId());
- killAndRestartLeader = killAndRestartServer(leader.getId(), 2000, 4000,
cluster, log);
- } else {
- killAndRestartLeader = CompletableFuture.completedFuture(null);
- }
-
- log.info(cluster.printServers());
+ CompletableFuture<Void> killAndRestartLeader =
CompletableFuture.completedFuture(null);
final SimpleMessage[] messages = SimpleMessage.create(numMessages);
- try (final RaftClient client = cluster.createClient()) {
- final AtomicInteger asyncReplyCount = new AtomicInteger();
- final CompletableFuture<Void> f = new CompletableFuture<>();
+ try {
+ log.info(cluster.printServers());
+
+ try (final RaftClient client = cluster.createClient()) {
+ final List<CompletableFuture<RaftClientReply>> asyncReplies = new
ArrayList<>();
- for (SimpleMessage message : messages) {
+ for (SimpleMessage message : messages) {
+ if (async) {
+ asyncReplies.add(client.async().send(message));
+ } else {
+ final RaftClientReply reply = client.io().send(message);
+ Assertions.assertTrue(reply.isSuccess());
+ }
+ }
if (async) {
- client.async().send(message).thenAcceptAsync(reply -> {
- if (!reply.isSuccess()) {
- f.completeExceptionally(
- new AssertionError("Failed with reply " + reply));
- } else if (asyncReplyCount.incrementAndGet() == messages.length) {
- f.complete(null);
- }
+ CompletableFuture.allOf(asyncReplies.toArray(new
CompletableFuture<?>[0])).join();
Review Comment:
Done. Removed the redundant `CompletableFuture.allOf(...).join()` and kept
the reply list cleanup.
--
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]
