feat: add TcpStackAwareSocketChannel for z/OS TCP/IP stack recovery — GH#4776

apiml-tomcat-common/src/main/java/org/zowe/apiml/product/web/TomcatAcceptFixConfig.java

@@ -18,6 +18,7 @@
 import org.apache.coyote.ProtocolHandler;
 import org.apache.tomcat.util.net.AbstractEndpoint;
 import org.apache.tomcat.util.net.NioEndpoint;
+import org.apache.tomcat.util.net.SocketWrapperBase;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.web.embedded.tomcat.TomcatConnectorCustomizer;
 import org.springframework.context.annotation.Bean;
@@ -54,8 +55,9 @@ public class TomcatAcceptFixConfig {
     private static final Field ENDPOINT_FIELD;
     private static final Field NIO_SOCKET_FIELD;
 
-    private static final MethodHandle IMPL_CLOSE_SELECTABGLE_CHANNEL_HANLE; // NOSONAR
+    private static final MethodHandle IMPL_CLOSE_SELECTABLE_CHANNEL_HANDLE; // NOSONAR
     private static final MethodHandle IMPL_CONFIGURE_BLOCKING; // NOSONAR
+    private static final String NETWORK_RECYCLED_EXCEPTION_CLASS = "com.ibm.net.NetworkRecycledException";
 
     /**
      * To mitigate parallel treatment of socket rebinding
@@ -69,7 +71,7 @@ public class TomcatAcceptFixConfig {
 
             Method implCloseSelectableChannel = AbstractSelectableChannel.class.getDeclaredMethod("implCloseSelectableChannel");
             implCloseSelectableChannel.setAccessible(true); // NOSONAR
-            IMPL_CLOSE_SELECTABGLE_CHANNEL_HANLE = MethodHandles.lookup().unreflect(implCloseSelectableChannel);
+            IMPL_CLOSE_SELECTABLE_CHANNEL_HANDLE = MethodHandles.lookup().unreflect(implCloseSelectableChannel);
 
             Method implConfigureBlocking = AbstractSelectableChannel.class.getDeclaredMethod("implConfigureBlocking", boolean.class);
             implConfigureBlocking.setAccessible(true); // NOSONAR
@@ -143,12 +145,32 @@ public void stopping() {
         running.set(false);
     }
 
+    static boolean isRecycledClass(Throwable t) {
+        return NETWORK_RECYCLED_EXCEPTION_CLASS.equals(t.getClass().getName());
+    }
+
+    static boolean isTcpStackRestarted(Throwable t) {
+        if ((t.getMessage() != null) && t.getMessage().contains("EDC5122I")) {
+            return true;
+        }
+
+        if (isRecycledClass(t)) {
+            return true;
+        }
+
+        Throwable cause = t.getCause();
+        if ((cause != null) && (cause != t)) {
+            return isTcpStackRestarted(cause);
+        }
+
+        return false;
+    }
+
     /**
      * Socket implementation wrapper to handle rebinding on TCP Stack restart
      */
     class FixedServerSocketChannel extends ServerSocketChannel {
 
-        private static final String NETWORK_RECYCLED_EXCEPTION_CLASS = "com.ibm.net.NetworkRecycledException";
 
         /**
          * Wrapper server socket inside
@@ -181,7 +203,7 @@ class FixedServerSocketChannel extends ServerSocketChannel {
         @Override
         protected void implCloseSelectableChannel() throws IOException {
             try {
-                IMPL_CLOSE_SELECTABGLE_CHANNEL_HANLE.invoke(socket);
+                IMPL_CLOSE_SELECTABLE_CHANNEL_HANDLE.invoke(socket);
             } catch (IOException | RuntimeException e) {
                 throw e;
             } catch (Throwable t) {
@@ -222,6 +244,21 @@ private void bindWithWait() throws IOException, InterruptedException {
             }
         }
 
+        private void closeConnectionsAfterTcpStackRestart() {
+            int closed = 0;
+            int failed = 0;
+            for (SocketWrapperBase<?> connection : abstractEndpoint.getConnections()) {
+                try {
+                    connection.close();
+                    closed++;
+                } catch (RuntimeException e) {
+                    failed++;
+                    log.debug("Unable to close a stale client connection after TCP/IP stack restart", e);
+                }
+            }
+            log.info("Closed {} client connection(s) after TCP/IP stack restart; {} could not be closed", closed, failed);
+        }
+
         /**
          * Rebind the server socket. The action could be done just by one thread. Other treats are waiting to be finish
          * by first one.
@@ -233,6 +270,8 @@ private void bindWithWait() throws IOException, InterruptedException {
         private synchronized void rebind(int stateBefore) throws IOException {
             if (state.compareAndSet(stateBefore, stateBefore + 1)) {
                 try {
+                    closeConnectionsAfterTcpStackRestart();
+
                     // socket must be closed before new binding
                     socket.close();
 
@@ -249,34 +288,13 @@ private synchronized void rebind(int stateBefore) throws IOException {
             }
         }
 
-        boolean isRecycledClass(Throwable t) {
-            return NETWORK_RECYCLED_EXCEPTION_CLASS.equals(t.getClass().getName());
-        }
-
-        boolean isTcpStackRestarted(Throwable t) {
-            if ((t.getMessage() != null) && t.getMessage().contains("EDC5122I")) {
-                return true;
-            }
-
-            if (isRecycledClass(t)) {
-                return true;
-            }
-
-            Throwable cause = t.getCause();
-            if ((cause != null) && (cause != t)) {
-                return isTcpStackRestarted(cause);
-            }
-
-            return false;
-        }
-
         public SocketChannel accept() throws IOException {
             // obtain current state of rebinding to detection parallel actions
             final int stateBefore = state.get();
             try {
                 return socket.accept();
             } catch (IOException ioe) {
-                if (isTcpStackRestarted(ioe)) {
+                if (TomcatAcceptFixConfig.isTcpStackRestarted(ioe)) {
                     // the fix solve just one issue about stopped TCP/IP stack
                     log.debug("The TCP/IP stack was probably restarted. The socket of Tomcat will rebind.");
                     rebind(stateBefore);
@@ -289,7 +307,7 @@ public SocketChannel accept() throws IOException {
     }
 
     /**
-     * The list of final methods, which cannot be delegated. See {@link FixedServerSocketChannel#socket}
+     * Methods final on {@link ServerSocketChannel} that Lombok must not delegate.
      */
     private interface Overridden {
 

apiml-tomcat-common/src/test/java/com/ibm/net/NetworkRecycledException.java

@@ -0,0 +1,14 @@
+/*
+ * This program and the accompanying materials are made available under the terms of the
+ * Eclipse Public License v2.0 which accompanies this distribution, and is available at
+ * https://www.eclipse.org/legal/epl-v20.html
+ *
+ * SPDX-License-Identifier: EPL-2.0
+ *
+ * Copyright Contributors to the Zowe Project.
+ */
+
+package com.ibm.net;
+
+public class NetworkRecycledException extends RuntimeException {
+}

apiml-tomcat-common/src/test/java/org/zowe/apiml/product/web/TomcatAcceptFixConfigTest.java

@@ -20,9 +20,11 @@
 import org.apache.coyote.http11.Http11NioProtocol;
 import org.apache.tomcat.util.net.AbstractEndpoint;
 import org.apache.tomcat.util.net.NioEndpoint;
+import org.apache.tomcat.util.net.SocketWrapperBase;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
+import com.ibm.net.NetworkRecycledException;
 import org.springframework.boot.web.embedded.tomcat.TomcatConnectorCustomizer;
 
 import java.io.IOException;
@@ -38,6 +40,7 @@
 import java.util.stream.Stream;
 
 import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.*;
 
 class TomcatAcceptFixConfigTest {
@@ -90,8 +93,10 @@
     }
 
     @Test
-    void givenCustomizedConnector_whenTcpipIsRestarted_thenRebind() throws Exception {
+    void givenCustomizedConnectorWithOpenConnections_whenTcpipIsRestarted_thenCloseConnectionsAndRebind() throws Exception {
         AtomicInteger counter = new AtomicInteger(0);
+        SocketWrapperBase<?> staleConnection = mock(SocketWrapperBase.class);
+        doReturn(Set.of(staleConnection)).when(testEndpoint).getConnections();
         doAnswer(invocation -> {
             if (counter.getAndIncrement() == 0) {
                 throw new IOException("EDC5122I");
@@ -101,6 +106,28 @@
         }).when(serverSocket).accept();
 
         assertSame(socketChannel, testEndpoint.serverSocketAccept());
+        verify(staleConnection, times(1)).close();
+        verify(serverSocket, times(1)).implCloseSelectableChannel();
+        verify(testEndpoint, times(1)).bind();
+    }
+
+    @Test
+    void givenOneConnectionCannotBeClosed_whenTcpipIsRestarted_thenCloseRemainingConnectionsAndRebind() throws Exception {
+        AtomicInteger counter = new AtomicInteger(0);
+        SocketWrapperBase<?> failingConnection = mock(SocketWrapperBase.class);
+        SocketWrapperBase<?> staleConnection = mock(SocketWrapperBase.class);
+        doReturn(Set.of(failingConnection, staleConnection)).when(testEndpoint).getConnections();
+        doThrow(new IllegalStateException("already closed")).when(failingConnection).close();
+        doAnswer(invocation -> {
+            if (counter.getAndIncrement() == 0) {
+                throw new IOException("EDC5122I");
+            }
+            return socketChannel;
+        }).when(serverSocket).accept();
+
+        assertSame(socketChannel, testEndpoint.serverSocketAccept());
+        verify(failingConnection, times(1)).close();
+        verify(staleConnection, times(1)).close();
         verify(serverSocket, times(1)).implCloseSelectableChannel();
         verify(testEndpoint, times(1)).bind();
     }
@@ -230,45 +257,40 @@
     @Nested
     class TcpStackRestartHandling {
 
-        ServerSocketChannel serverSocket = new TestServerSocketChannel(mock(SelectorProvider.class));
-        TomcatAcceptFixConfig.FixedServerSocketChannel channel = new TomcatAcceptFixConfig().new FixedServerSocketChannel(serverSocket, null, null);
-
         @Test
         void givenExceptionWithTheMessage_whenHandle_thenReturnTrue() {
-            assertTrue(channel.isTcpStackRestarted(new RuntimeException("EDC5122I TCP Stack restarted")));
+            assertTrue(TomcatAcceptFixConfig.isTcpStackRestarted(new RuntimeException("EDC5122I TCP Stack restarted")));
         }
 
         @Test
         void givenExceptionWithCyclicCause_whenHandle_thenReturnFalse() {
             Exception e = spy(new RuntimeException("Error"));
             doReturn(e).when(e).getCause();
-            assertFalse(channel.isTcpStackRestarted(e));
+            assertFalse(TomcatAcceptFixConfig.isTcpStackRestarted(e));
         }
 
         @Test
         void givenExceptionWithTheMessageAsCause_whenHandle_thenReturnTrue() {
             Exception e = new RuntimeException("EDC5122I TCP Stack restarted");
             e = new RuntimeException("Wrapper1", e);
             e = new RuntimeException("Wrapper2", e);
-            assertTrue(channel.isTcpStackRestarted(e));
+            assertTrue(TomcatAcceptFixConfig.isTcpStackRestarted(e));
         }
 
         @Test
-        void givenExceptionWithSpecificClassName_whenHandle_thenReturnTrue() {
-            TomcatAcceptFixConfig.FixedServerSocketChannel channel = new TomcatAcceptFixConfig().new FixedServerSocketChannel(serverSocket, null, null) {
-                @Override
-                boolean isRecycledClass(Throwable t) {
-                    return "java.lang.IllegalArgumentException".equals(t.getClass().getName());
-                }
-            };
-
-            Exception e = new IllegalArgumentException("Tested exception");
+        void givenNetworkRecycledException_whenIsRecycledClass_thenReturnTrue() {
+            Exception e = new NetworkRecycledException();
             e = new RuntimeException("Wrapper", e);
-            assertTrue(channel.isTcpStackRestarted(e));
+
+            assertTrue(TomcatAcceptFixConfig.isTcpStackRestarted(e));
         }
 
-    }
+        @Test
+        void givenNonMatchingException_whenIsRecycledClass_thenReturnFalse() {
+            assertFalse(TomcatAcceptFixConfig.isRecycledClass(new RuntimeException("not a match")));
+        }
 
+    }
     private static class TestEndpoint extends NioEndpoint {
 
         public TestEndpoint(ServerSocketChannel serverSocket) {
@@ -363,4 +385,4 @@
 
     }
 
-}
+}