You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ib...@apache.org on 2024/02/09 15:02:32 UTC
(ignite-3) branch main updated: IGNITE-21493 Apply ack silencer when the channel is open (#3196)
This is an automated email from the ASF dual-hosted git repository.
ibessonov pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 4b3e9bf163 IGNITE-21493 Apply ack silencer when the channel is open (#3196)
4b3e9bf163 is described below
commit 4b3e9bf16316cfeca40cf83a4e748202c26a6382
Author: Roman Puchkovskiy <ro...@gmail.com>
AuthorDate: Fri Feb 9 19:02:26 2024 +0400
IGNITE-21493 Apply ack silencer when the channel is open (#3196)
---
.../network/netty/ItConnectionManagerTest.java | 2 ++
.../scalecube/ItScaleCubeNetworkMessagingTest.java | 20 +++++++++++++++-----
2 files changed, 17 insertions(+), 5 deletions(-)
diff --git a/modules/network/src/integrationTest/java/org/apache/ignite/internal/network/netty/ItConnectionManagerTest.java b/modules/network/src/integrationTest/java/org/apache/ignite/internal/network/netty/ItConnectionManagerTest.java
index 0b5254c7fb..c4cd4f751c 100644
--- a/modules/network/src/integrationTest/java/org/apache/ignite/internal/network/netty/ItConnectionManagerTest.java
+++ b/modules/network/src/integrationTest/java/org/apache/ignite/internal/network/netty/ItConnectionManagerTest.java
@@ -427,6 +427,7 @@ public class ItConnectionManagerTest extends BaseIgniteAbstractTest {
ConnectionManagerWrapper manager2 = startManager(4001)
) {
NettySender sender = manager1.openChannelTo(manager2).toCompletableFuture().get(10, TimeUnit.SECONDS);
+ waitTillChannelAppearsInMapOnAcceptor(sender, manager1, manager2);
OutgoingAcknowledgementSilencer ackSilencer = dropAcksFrom(manager2);
@@ -453,6 +454,7 @@ public class ItConnectionManagerTest extends BaseIgniteAbstractTest {
ConnectionManagerWrapper manager2 = startManager(4001)
) {
NettySender sender = manager1.openChannelTo(manager2).toCompletableFuture().get(10, TimeUnit.SECONDS);
+ waitTillChannelAppearsInMapOnAcceptor(sender, manager1, manager2);
dropAcksFrom(manager2);
diff --git a/modules/network/src/integrationTest/java/org/apache/ignite/internal/network/scalecube/ItScaleCubeNetworkMessagingTest.java b/modules/network/src/integrationTest/java/org/apache/ignite/internal/network/scalecube/ItScaleCubeNetworkMessagingTest.java
index cd6ff9ca76..df80f7e3ce 100644
--- a/modules/network/src/integrationTest/java/org/apache/ignite/internal/network/scalecube/ItScaleCubeNetworkMessagingTest.java
+++ b/modules/network/src/integrationTest/java/org/apache/ignite/internal/network/scalecube/ItScaleCubeNetworkMessagingTest.java
@@ -959,7 +959,7 @@ class ItScaleCubeNetworkMessagingTest {
// Open a channel to allow a silencer to be installed on it.
openDefaultChannelBetween(sender, receiver);
- OutgoingAcknowledgementSilencer ackSilencer = dropAcksFrom(receiver);
+ OutgoingAcknowledgementSilencer ackSilencer = dropAcksWhenDefaultChannelOpens(receiver);
CompletableFuture<Void> sendFuture = operation.send(
sender.messagingService(),
@@ -996,10 +996,20 @@ class ItScaleCubeNetworkMessagingTest {
return send(messageFactory.testMessage().build(), sender, receiver);
}
- private static OutgoingAcknowledgementSilencer dropAcksFrom(ClusterService clusterService) throws InterruptedException {
+ private static OutgoingAcknowledgementSilencer dropAcksWhenDefaultChannelOpens(ClusterService clusterService)
+ throws InterruptedException {
DefaultMessagingService messagingService = (DefaultMessagingService) clusterService.messagingService();
- return OutgoingAcknowledgementSilencer.installOn(messagingService.connectionManager().channels().values());
+ ConnectionManager connectionManager = messagingService.connectionManager();
+ assertTrue(
+ waitForCondition(
+ () -> connectionManager.channels().keySet().stream().anyMatch(key -> key.type() == ChannelType.DEFAULT),
+ SECONDS.toMillis(10)
+ ),
+ "Did not see a default channel to be opened on the connection manager"
+ );
+
+ return OutgoingAcknowledgementSilencer.installOn(connectionManager.channels().values());
}
private void provokeAckFor(ClusterService sideToGetAck, ClusterService sideToSendAck) {
@@ -1019,7 +1029,7 @@ class ItScaleCubeNetworkMessagingTest {
echoMessagesBackAt(outcast);
openDefaultChannelBetween(notOutcast, outcast);
- dropAcksFrom(outcast);
+ dropAcksWhenDefaultChannelOpens(outcast);
CompletableFuture<Void> sendFuture = operation.send(
notOutcast.messagingService(),
@@ -1045,7 +1055,7 @@ class ItScaleCubeNetworkMessagingTest {
echoMessagesBackAt(receiver);
openDefaultChannelBetween(sender, receiver);
- dropAcksFrom(receiver);
+ dropAcksWhenDefaultChannelOpens(receiver);
CompletableFuture<Void> sendFuture = operation.send(
sender.messagingService(),