IGNITE-29072 Fix tests using Thread.suspend() for JDK 21

modules/core/src/test/java/org/apache/ignite/internal/IgniteClientConnectAfterCommunicationFailureTest.java

@@ -25,6 +25,7 @@
 import org.apache.ignite.configuration.IgniteConfiguration;
 import org.apache.ignite.internal.util.nio.GridCommunicationClient;
 import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.spi.discovery.tcp.TestTcpDiscoverySpi;
 import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
 import org.junit.Test;
 
@@ -66,20 +67,15 @@ public void testClientThreadsSuspended() throws Exception {
         Ignite srv2 = startGrid("server2");
         Ignite client = startClientGrid("client");
 
-        boolean blockedAnything = false;
+        TestTcpDiscoverySpi clientSpi = (TestTcpDiscoverySpi)client.configuration().getDiscoverySpi();
 
-        for (Thread thread : Thread.getAllStackTraces().keySet()) {
-            if (thread.getName().contains("%client%")) {
-                thread.suspend();
-                blockedAnything = true;
-            }
-        }
+        clientSpi.freeze();
 
-        Thread.sleep(10000);
-
-        for (Thread thread : Thread.getAllStackTraces().keySet()) {
-            if (thread.getName().contains("%client%"))
-                thread.resume();
+        try {
+            Thread.sleep(10000);
+        }
+        finally {
+            clientSpi.unfreeze();
         }
 
         for (int j = 0; j < 10; j++) {
@@ -102,7 +98,6 @@ public void testClientThreadsSuspended() throws Exception {
                 Thread.sleep(1000);
         }
 
-        assertTrue(blockedAnything);
         assertEquals(1, srv2.cluster().forClients().nodes().size());
         assertEquals(1, srv1.cluster().forClients().nodes().size());
     }

modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpClientDiscoverySpiSelfTest.java

@@ -63,7 +63,6 @@
 import org.apache.ignite.resources.IgniteInstanceResource;
 import org.apache.ignite.spi.IgniteSpiException;
 import org.apache.ignite.spi.IgniteSpiOperationTimeoutHelper;
-import org.apache.ignite.spi.IgniteSpiThread;
 import org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi;
 import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryNode;
 import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
@@ -2486,6 +2485,9 @@ protected static class TestTcpDiscoverySpi extends TcpDiscoverySpi {
         /** */
         private final AtomicBoolean openSockLock = new AtomicBoolean();
 
+        /** */
+        private final AtomicBoolean readLock = new AtomicBoolean();
+
         /** */
         private AtomicInteger failNodeAdded = new AtomicInteger();
 
@@ -2662,33 +2664,41 @@ public void pauseSocketWrite() {
         }
 
         /**
-         * @param suspend If {@code true} suspends worker threads.
+         * @param pauseRead If {@code true} also pauses socket reads, which emulates hanging worker threads.
          */
-        public void pauseAll(boolean suspend) {
-            pauseResumeOperation(true, openSockLock, writeLock);
-
-            if (suspend) {
-                for (Thread t : impl.threads())
-                    t.suspend();
-            }
+        public void pauseAll(boolean pauseRead) {
+            if (pauseRead)
+                pauseResumeOperation(true, openSockLock, writeLock, readLock);
+            else
+                pauseResumeOperation(true, openSockLock, writeLock);
         }
 
         /**
          *
          */
         public void resumeAll() {
-            pauseResumeOperation(false, openSockLock, writeLock);
-
-            for (IgniteSpiThread t : impl.threads())
-                t.resume();
+            pauseResumeOperation(false, openSockLock, writeLock, readLock);
         }
 
         /** {@inheritDoc} */
         @Override protected <T extends Message> T readMessage(
             TcpDiscoveryIoSession ses,
             long timeout
         ) throws IOException, IgniteCheckedException {
-            return msgTracker.track(ses, super.readMessage(ses, timeout));
+            waitFor(readLock);
+
+            T msg;
+
+            try {
+                msg = super.readMessage(ses, timeout);
+            }
+            finally {
+                // A reader may have been blocked on the socket before the pause, so hold the result (a message or
+                // a failure) until resume.
+                waitFor(readLock);
+            }
+
+            return msgTracker.track(ses, msg);
         }
 
         /** {@inheritDoc} */
@@ -2723,7 +2733,16 @@ public void resumeAll() {
 
         /** {@inheritDoc} */
         @Override protected int readReceipt(TcpDiscoveryIoSession ses, long timeout) throws IOException {
-            int res = super.readReceipt(ses, timeout);
+            waitFor(readLock);
+
+            int res;
+
+            try {
+                res = super.readReceipt(ses, timeout);
+            }
+            finally {
+                waitFor(readLock);
+            }
 
             if (res != TcpDiscoveryImpl.RES_OK) {
                 invalidRes = true;

modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryNetworkIssuesTest.java

@@ -34,7 +34,6 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.BiConsumer;
-import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import org.apache.ignite.Ignite;
 import org.apache.ignite.IgniteCheckedException;
@@ -47,12 +46,7 @@
 import org.apache.ignite.internal.util.GridConcurrentHashSet;
 import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.G;
-import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.internal.util.worker.GridWorker;
 import org.apache.ignite.spi.IgniteSpiOperationTimeoutHelper;
-import org.apache.ignite.spi.communication.CommunicationSpi;
-import org.apache.ignite.spi.communication.tcp.internal.GridNioServerWrapper;
-import org.apache.ignite.spi.discovery.DiscoverySpi;
 import org.apache.ignite.spi.discovery.tcp.internal.TcpDiscoveryNode;
 import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
@@ -141,7 +135,7 @@ public class TcpDiscoveryNetworkIssuesTest extends GridCommonAbstractTest {
     @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
         IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
 
-        TcpDiscoverySpi spi = (specialSpi != null) ? specialSpi : new TcpDiscoverySpi();
+        TcpDiscoverySpi spi = (specialSpi != null) ? specialSpi : new TestTcpDiscoverySpi();
 
         if (usePortFromNodeName)
             spi.setLocalPort(Integer.parseInt(igniteInstanceName.split("-")[1]));
@@ -471,13 +465,13 @@ private void simulateFailureOfTwoNodes(boolean sequentionally) throws Exception
         else
             failedNodes.add(4);
 
-        failedNodes.forEach(idx -> processNetworkThreads(ignite(idx), Thread::suspend));
+        failedNodes.forEach(idx -> ((TestTcpDiscoverySpi)spi(ignite(idx))).freeze());
 
         try {
             failLatch.await(10, TimeUnit.SECONDS);
         }
         finally {
-            failedNodes.forEach(idx -> processNetworkThreads(ignite(idx), Thread::resume));
+            failedNodes.forEach(idx -> ((TestTcpDiscoverySpi)spi(ignite(idx))).unfreeze());
         }
 
         for (int i = 0; i < gridCnt; i++) {
@@ -547,25 +541,6 @@ private void breakDiscoConnectionToNext(IgniteEx ig) throws Exception {
         ses.socket().getOutputStream().close();
     }
 
-    /**
-     * Simulates network failure on certain node.
-     */
-    private void processNetworkThreads(Ignite ignite, Consumer<Thread> proc) {
-        DiscoverySpi disco = ignite.configuration().getDiscoverySpi();
-
-        ServerImpl serverImpl = U.field(disco, "impl");
-
-        for (Thread thread : serverImpl.threads())
-            proc.accept(thread);
-
-        CommunicationSpi<?> comm = ignite.configuration().getCommunicationSpi();
-
-        GridNioServerWrapper nioServerWrapper = U.field(comm, "nioSrvWrapper");
-
-        for (GridWorker worker : nioServerWrapper.nio().workers())
-            proc.accept(worker.runner());
-    }
-
     /** */
     private static TestDiscoverySpi testSpi(Ignite ig) {
         return ((TestDiscoverySpi)ig.configuration().getDiscoverySpi());

modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TestTcpDiscoverySpi.java

@@ -21,14 +21,17 @@
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.InterruptedIOException;
 import java.io.OutputStream;
 import java.net.Socket;
 import java.util.Arrays;
+import java.util.concurrent.CountDownLatch;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.internal.GridKernalContext;
 import org.apache.ignite.internal.managers.discovery.IgniteDiscoverySpiInternalListener;
 import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.plugin.extensions.communication.Message;
 import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage;
 import org.apache.ignite.spi.discovery.DiscoverySpiListener;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
@@ -53,9 +56,96 @@ public class TestTcpDiscoverySpi extends TcpDiscoverySpi implements IgniteDiscov
     /** */
     private IgniteDiscoverySpiInternalListener internalLsnr;
 
+    /** Latch released on {@link #unfreeze()}, {@code null} if the discovery I/O is not frozen. */
+    private volatile CountDownLatch freezeLatch;
+
+    /**
+     * Freezes the discovery I/O of this node: every socket read and write blocks until {@link #unfreeze()} is called.
+     * The node keeps accepting TCP connections. Emulates a node whose threads hang, e.g. at a long GC pause.
+     */
+    public synchronized void freeze() {
+        if (freezeLatch == null)
+            freezeLatch = new CountDownLatch(1);
+    }
+
+    /** Releases the discovery I/O frozen by {@link #freeze()}. */
+    public synchronized void unfreeze() {
+        if (freezeLatch != null) {
+            freezeLatch.countDown();
+
+            freezeLatch = null;
+        }
+    }
+
+    /**
+     * Blocks while the discovery I/O is frozen.
+     *
+     * @throws InterruptedIOException If interrupted.
+     */
+    private void awaitUnfrozen() throws InterruptedIOException {
+        CountDownLatch latch = freezeLatch;
+
+        if (latch == null)
+            return;
+
+        try {
+            latch.await();
+        }
+        catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+
+            throw new InterruptedIOException("Interrupted while discovery I/O is frozen.");
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void write(TcpDiscoveryIoSession ses, byte[] data, long timeout) throws IOException,
+        IgniteCheckedException {
+        awaitUnfrozen();
+
+        super.write(ses, data, timeout);
+    }
+
+    /** {@inheritDoc} */
+    @Override protected void writeReceipt(TcpDiscoveryIoSession ses, int res, long timeout) throws IOException,
+        IgniteCheckedException {
+        awaitUnfrozen();
+
+        super.writeReceipt(ses, res, timeout);
+    }
+
+    /** {@inheritDoc} */
+    @Override protected <T extends Message> T readMessage(TcpDiscoveryIoSession ses, long timeout) throws IOException,
+        IgniteCheckedException {
+        awaitUnfrozen();
+
+        try {
+            return super.readMessage(ses, timeout);
+        }
+        finally {
+            // A reader may have been blocked on the socket before the freeze, so hold the result (a message or
+            // a failure) until unfreeze.
+            awaitUnfrozen();
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override protected int readReceipt(TcpDiscoveryIoSession ses, long timeout) throws IOException {
+        awaitUnfrozen();
+
+        try {
+            return super.readReceipt(ses, timeout);
+        }
+        finally {
+            awaitUnfrozen();
+        }
+    }
+
     /** {@inheritDoc} */
     @Override protected void writeMessage(TcpDiscoveryIoSession ses, TcpDiscoveryAbstractMessage msg, long timeout) throws IOException,
         IgniteCheckedException {
+        awaitUnfrozen();
+
         if (msg instanceof TcpDiscoveryPingResponse && ignorePingResponse)
             return;