You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ap...@apache.org on 2024/02/05 14:43:53 UTC
(ignite-3) branch main updated: IGNITE-21442 DefaultMessagingServiceTest.sendMessagesTwoChannels() hangs (#3154)
This is an automated email from the ASF dual-hosted git repository.
apolovtsev 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 04b732e282 IGNITE-21442 DefaultMessagingServiceTest.sendMessagesTwoChannels() hangs (#3154)
04b732e282 is described below
commit 04b732e2826d7613b41c9d5d9af7ec7071de04d9
Author: Roman Puchkovskiy <ro...@gmail.com>
AuthorDate: Mon Feb 5 18:43:47 2024 +0400
IGNITE-21442 DefaultMessagingServiceTest.sendMessagesTwoChannels() hangs (#3154)
---
.../internal/network/NettyWorkersRegistrar.java | 10 +--
.../internal/network/netty/ConnectionManager.java | 2 +-
.../network/DefaultMessagingServiceTest.java | 76 +++++++++++++---------
3 files changed, 54 insertions(+), 34 deletions(-)
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/NettyWorkersRegistrar.java b/modules/network/src/main/java/org/apache/ignite/internal/network/NettyWorkersRegistrar.java
index c7711363c1..6ad0091041 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/NettyWorkersRegistrar.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/NettyWorkersRegistrar.java
@@ -26,7 +26,6 @@ import io.netty.util.concurrent.SingleThreadEventExecutor;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
@@ -55,7 +54,7 @@ public class NettyWorkersRegistrar implements IgniteComponent {
private final CriticalWorkersConfiguration criticalWorkersConfiguration;
- private volatile List<NettyWorker> workers = new CopyOnWriteArrayList<>();
+ private volatile List<NettyWorker> workers;
@Nullable
private volatile ScheduledFuture<?> sendHearbeatsTaskFuture;
@@ -122,8 +121,11 @@ public class NettyWorkersRegistrar implements IgniteComponent {
heartBeatsTaskFuture.cancel(false);
}
- for (NettyWorker worker : workers) {
- criticalWorkerRegistry.unregister(worker);
+ List<NettyWorker> registeredWorkers = workers;
+ if (registeredWorkers != null) {
+ for (NettyWorker worker : registeredWorkers) {
+ criticalWorkerRegistry.unregister(worker);
+ }
}
}
diff --git a/modules/network/src/main/java/org/apache/ignite/internal/network/netty/ConnectionManager.java b/modules/network/src/main/java/org/apache/ignite/internal/network/netty/ConnectionManager.java
index eebb66d7dc..6549418159 100644
--- a/modules/network/src/main/java/org/apache/ignite/internal/network/netty/ConnectionManager.java
+++ b/modules/network/src/main/java/org/apache/ignite/internal/network/netty/ConnectionManager.java
@@ -448,7 +448,7 @@ public class ConnectionManager implements ChannelCreationListener {
CompletableFuture<Void> finalStopFuture = allOf(stopFutures.toArray(CompletableFuture<?>[]::new));
try {
- finalStopFuture.join();
+ finalStopFuture.get();
} catch (Exception e) {
LOG.warn("Failed to stop connection manager [reason={}]", e.getMessage());
}
diff --git a/modules/network/src/test/java/org/apache/ignite/internal/network/DefaultMessagingServiceTest.java b/modules/network/src/test/java/org/apache/ignite/internal/network/DefaultMessagingServiceTest.java
index c3ffa23cf4..2c8f624944 100644
--- a/modules/network/src/test/java/org/apache/ignite/internal/network/DefaultMessagingServiceTest.java
+++ b/modules/network/src/test/java/org/apache/ignite/internal/network/DefaultMessagingServiceTest.java
@@ -184,18 +184,22 @@ class DefaultMessagingServiceTest extends BaseIgniteAbstractTest {
mockSerializationRegistry(longWaitSerializer));
Services receiverServices = createMessagingService(receiverNode, receiverNetworkConfig)
) {
- CountDownLatch latch = new CountDownLatch(2);
- receiverServices.messagingService.addMessageHandler(
- TestMessageTypes.class,
- (message, sender, correlationId) -> latch.countDown()
- );
-
- senderServices.messagingService.send(receiverNode, TestMessageImpl.builder().build());
- senderServices.messagingService.send(receiverNode, AllTypesMessageImpl.builder().build());
-
- assertThat(latch.getCount(), is(2L));
- release.set(true);
- assertTrue(latch.await(1, TimeUnit.SECONDS));
+ try {
+ CountDownLatch latch = new CountDownLatch(2);
+ receiverServices.messagingService.addMessageHandler(
+ TestMessageTypes.class,
+ (message, sender, correlationId) -> latch.countDown()
+ );
+
+ senderServices.messagingService.send(receiverNode, TestMessageImpl.builder().build());
+ senderServices.messagingService.send(receiverNode, AllTypesMessageImpl.builder().build());
+
+ assertThat(latch.getCount(), is(2L));
+ release.set(true);
+ assertTrue(latch.await(1, TimeUnit.SECONDS));
+ } finally {
+ release.set(true);
+ }
}
}
@@ -215,20 +219,24 @@ class DefaultMessagingServiceTest extends BaseIgniteAbstractTest {
mockSerializationRegistry(longWaitSerializer));
Services receiverServices = createMessagingService(receiverNode, receiverNetworkConfig)
) {
- CountDownLatch latch = new CountDownLatch(2);
- receiverServices.messagingService.addMessageHandler(
- TestMessageTypes.class,
- (message, sender, correlationId) -> latch.countDown()
- );
-
- senderServices.messagingService.send(receiverNode, TestMessageImpl.builder().build());
- senderServices.messagingService.send(receiverNode, TEST_CHANNEL, AllTypesMessageImpl.builder().build());
-
- await().timeout(1, TimeUnit.SECONDS)
- .until(() -> latch.getCount() == 1);
-
- release.set(true);
- assertTrue(latch.await(1, TimeUnit.SECONDS));
+ try {
+ CountDownLatch latch = new CountDownLatch(2);
+ receiverServices.messagingService.addMessageHandler(
+ TestMessageTypes.class,
+ (message, sender, correlationId) -> latch.countDown()
+ );
+
+ senderServices.messagingService.send(receiverNode, TestMessageImpl.builder().build());
+ senderServices.messagingService.send(receiverNode, TEST_CHANNEL, AllTypesMessageImpl.builder().build());
+
+ await().timeout(10, TimeUnit.SECONDS)
+ .until(() -> latch.getCount() == 1);
+
+ release.set(true);
+ assertTrue(latch.await(1, TimeUnit.SECONDS));
+ } finally {
+ release.set(true);
+ }
}
}
@@ -482,7 +490,7 @@ class DefaultMessagingServiceTest extends BaseIgniteAbstractTest {
messagingService.setConnectionManager(connectionManager);
- return new Services(connectionManager, messagingService);
+ return new Services(connectionManager, messagingService, bootstrapFactory);
}
private static RecoveryClientHandshakeManagerFactory clientHandshakeManagerFactoryAdding(Runnable beforeHandshake,
@@ -519,15 +527,25 @@ class DefaultMessagingServiceTest extends BaseIgniteAbstractTest {
private static class Services implements AutoCloseable {
private final ConnectionManager connectionManager;
private final DefaultMessagingService messagingService;
+ private final NettyBootstrapFactory bootstrapFactory;
- private Services(ConnectionManager connectionManager, DefaultMessagingService messagingService) {
+ private Services(
+ ConnectionManager connectionManager,
+ DefaultMessagingService messagingService,
+ NettyBootstrapFactory bootstrapFactory
+ ) {
this.connectionManager = connectionManager;
this.messagingService = messagingService;
+ this.bootstrapFactory = bootstrapFactory;
}
@Override
public void close() throws Exception {
- IgniteUtils.closeAll(connectionManager::initiateStopping, connectionManager::stop, messagingService::stop);
+ IgniteUtils.closeAll(
+ connectionManager::initiateStopping, connectionManager::stop,
+ messagingService::stop,
+ bootstrapFactory::beforeNodeStop, bootstrapFactory::stop
+ );
}
}