Trigger connection recovery on socket write error

Automatic connection recovery triggers now by default when a write
operation fails because of an IO exception. The recovery process
takes place in a dedicated thread so the write operation doesn't
wait (it receives the same IO exception immediatly).

The test to trigger the error has changed: it doesn't use manual
ack anymore, as this could sometimes block the broker and make
recovery fail (broker was busy re-enqueuing messages). The test
now sends a message in the consumer, which is enough to reproduce
the error.

Note the test against NIO is skipped right now, as it needs
additional care.

[#154263515]

References #341

Author: acogoluegnes

src/main/java/com/rabbitmq/client/impl/AMQConnection.java

@@ -254,7 +254,7 @@ public AMQConnection(ConnectionParams params, FrameHandler frameHandler, Metrics
         this.errorOnWriteListener = params.getErrorOnWriteListener() != null ? params.getErrorOnWriteListener() :
             new ErrorOnWriteListener() {
                 @Override
-                public void handle(Connection connection, Throwable exception) { }
+                public void handle(Connection connection, IOException exception) { }
             };
 
     }
@@ -572,7 +572,6 @@ public void flush() throws IOException {
             _frameHandler.flush();
         } catch (IOException ioe) {
             this.errorOnWriteListener.handle(this, ioe);
-            throw ioe;
         }
     }
 
@@ -592,15 +591,24 @@ private class MainLoop implements Runnable {
          */
         @Override
         public void run() {
+            boolean shouldDoFinalShutdown = true;
             try {
                 while (_running) {
                     Frame frame = _frameHandler.readFrame();
                     readFrame(frame);
                 }
             } catch (Throwable ex) {
-                handleFailure(ex);
+                if (ex instanceof InterruptedException) {
+                    // loop has been interrupted during shutdown,
+                    // no need to do it again
+                    shouldDoFinalShutdown = false;
+                } else {
+                    handleFailure(ex);
+                }
             } finally {
-                doFinalShutdown();
+                if (shouldDoFinalShutdown) {
+                    doFinalShutdown();
+                }
             }
         }
     }
@@ -706,6 +714,7 @@ public void doFinalShutdown() {
         if (finalShutdownStarted.compareAndSet(false, true)) {
             _frameHandler.close();
             _appContinuation.set(null);
+            closeMainLoopThreadIfNecessary();
             notifyListeners();
             // assuming that shutdown listeners do not do anything
             // asynchronously, e.g. start new threads, this effectively
@@ -715,6 +724,22 @@ public void doFinalShutdown() {
         }
     }
 
+    private void closeMainLoopThreadIfNecessary() {
+        if (mainLoopReadThreadNotNull() && notInMainLoopThread()) {
+            if (this.mainLoopThread.isAlive()) {
+                this.mainLoopThread.interrupt();
+            }
+        }
+    }
+
+    private boolean notInMainLoopThread() {
+        return Thread.currentThread() != this.mainLoopThread;
+    }
+
+    private boolean mainLoopReadThreadNotNull() {
+        return this.mainLoopThread != null;
+    }
+
     private void notifyRecoveryCanBeginListeners() {
         ShutdownSignalException sse = this.getCloseReason();
         for(RecoveryCanBeginListener fn : Utility.copy(this.recoveryCanBeginListeners)) {

src/main/java/com/rabbitmq/client/impl/ErrorOnWriteListener.java

@@ -17,6 +17,8 @@
 
 import com.rabbitmq.client.Connection;
 
+import java.io.IOException;
+
 /**
  * Listener called when a connection gets an error trying to write on the socket.
  * This can be used to trigger connection recovery.
@@ -28,6 +30,6 @@ public interface ErrorOnWriteListener {
      * @param connection the owning connection instance
      * @param exception the thrown exception
      */
-    void handle(Connection connection, Throwable exception);
+    void handle(Connection connection, IOException exception) throws IOException;
 
 }

src/main/java/com/rabbitmq/client/impl/recovery/AutorecoveringConnection.java

@@ -22,13 +22,18 @@
 import com.rabbitmq.client.impl.FrameHandlerFactory;
 import com.rabbitmq.client.impl.NetworkConnection;
 import com.rabbitmq.utility.Utility;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.net.InetAddress;
 import java.util.*;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 
 /**
  * Connection implementation that performs automatic recovery when
@@ -52,6 +57,9 @@
  * @since 3.3.0
  */
 public class AutorecoveringConnection implements RecoverableConnection, NetworkConnection {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(AutorecoveringConnection.class);
+
     private final RecoveryAwareAMQConnectionFactory cf;
     private final Map<Integer, AutorecoveringChannel> channels;
     private final ConnectionParams params;
@@ -88,15 +96,37 @@ public AutorecoveringConnection(ConnectionParams params, FrameHandlerFactory f,
         this.cf = new RecoveryAwareAMQConnectionFactory(params, f, addressResolver, metricsCollector);
         this.params = params;
 
+        setupErrorOnWriteListenerForPotentialRecovery();
+
+        this.channels = new ConcurrentHashMap<Integer, AutorecoveringChannel>();
+    }
+
+    private void setupErrorOnWriteListenerForPotentialRecovery() {
+        final ThreadFactory threadFactory = this.params.getThreadFactory();
+        final Lock errorOnWriteLock = new ReentrantLock();
         this.params.setErrorOnWriteListener(new ErrorOnWriteListener() {
             @Override
-            public void handle(Connection connection, Throwable exception) {
-                AMQConnection c = (AMQConnection) connection;
-                c.handleIoError(exception);
+            public void handle(final Connection connection, final IOException exception) throws IOException {
+                // this is called for any write error
+                // we should trigger the error handling and the recovery only once
+                if (errorOnWriteLock.tryLock()) {
+                    try {
+                        Thread recoveryThread = threadFactory.newThread(new Runnable() {
+                            @Override
+                            public void run() {
+                                AMQConnection c = (AMQConnection) connection;
+                                c.handleIoError(exception);
+                            }
+                        });
+                        recoveryThread.setName("RabbitMQ Error On Write Thread");
+                        recoveryThread.start();
+                    } finally {
+                        errorOnWriteLock.unlock();
+                    }
+                }
+                throw exception;
             }
         });
-
-        this.channels = new ConcurrentHashMap<Integer, AutorecoveringChannel>();
     }
 
     /**
@@ -661,7 +691,9 @@ private void recoverConsumers() {
         for (Map.Entry<String, RecordedConsumer> entry : Utility.copy(this.consumers).entrySet()) {
             String tag = entry.getKey();
             RecordedConsumer consumer = entry.getValue();
-
+            if (LOGGER.isDebugEnabled()) {
+                LOGGER.debug("Recovering consumer {}", consumer);
+            }
             try {
                 String newTag = consumer.recover();
                 // make sure server-generated tags are re-added. MK.
@@ -676,6 +708,9 @@ private void recoverConsumers() {
                 for(ConsumerRecoveryListener crl : Utility.copy(this.consumerRecoveryListeners)) {
                     crl.consumerRecovered(tag, newTag);
                 }
+                if (LOGGER.isDebugEnabled()) {
+                    LOGGER.debug("Consumer {} has recovered", consumer);
+                }
             } catch (Exception cause) {
                 final String message = "Caught an exception while recovering consumer " + tag +
                         ": " + cause.getMessage();

src/test/java/com/rabbitmq/client/test/NoAutoRecoveryWhenTcpWindowIsFullTest.java

@@ -20,6 +20,7 @@
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.DefaultConsumer;
+import com.rabbitmq.client.DefaultSocketConfigurator;
 import com.rabbitmq.client.Envelope;
 import com.rabbitmq.client.Recoverable;
 import com.rabbitmq.client.RecoveryListener;
@@ -30,6 +31,7 @@
 import org.junit.Test;
 
 import java.io.IOException;
+import java.net.Socket;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
@@ -45,68 +47,71 @@
 /**
  * Test to trigger and check the fix of https://github.com/rabbitmq/rabbitmq-java-client/issues/341.
  * Conditions:
- * - client registers consumer and a call QoS after
- * - client get many messages and the consumer is slow
+ * - client registers consumer
+ * - client get many messages as the consumer is slow
  * - the work pool queue is full, the reading thread is stuck
  * - more messages come from the network and saturates the TCP buffer
  * - the connection dies but the client doesn't detect it
- * - acks of messages fail
+ * - sending in the consumer fails
  * - connection recovery is never triggered
  * <p>
  * The fix consists in triggering connection recovery when writing
- * to the socket fails. As the socket is dead, the closing
- * sequence can take some time, hence the setup of the shutdown
- * listener, which avoids waiting for the socket termination by
- * the OS (can take 15 minutes on linux).
+ * to the socket fails.
  */
 public class NoAutoRecoveryWhenTcpWindowIsFullTest {
 
-    private static final int QOS_PREFETCH = 64;
-    private static final int NUM_MESSAGES_TO_PRODUCE = 100000;
+    private static final int NUM_MESSAGES_TO_PRODUCE = 50000;
     private static final int MESSAGE_PROCESSING_TIME_MS = 3000;
+    private static final byte[] MESSAGE_CONTENT = ("MESSAGE CONTENT " + NUM_MESSAGES_TO_PRODUCE).getBytes();
 
-    private ExecutorService dispatchingService;
-    private ExecutorService producerService;
-    private ExecutorService shutdownService;
+    private ExecutorService executorService;
     private AutorecoveringConnection producingConnection;
     private AutorecoveringChannel producingChannel;
     private AutorecoveringConnection consumingConnection;
     private AutorecoveringChannel consumingChannel;
 
-    private CountDownLatch consumerRecoverOkLatch;
+    private CountDownLatch consumerOkLatch;
 
     @Before
     public void setUp() throws Exception {
-        dispatchingService = Executors.newSingleThreadExecutor();
-        shutdownService = Executors.newSingleThreadExecutor();
-        producerService = Executors.newSingleThreadExecutor();
+        // we need several threads to publish, dispatch deliveries, handle RPC responses, etc.
+        executorService = Executors.newFixedThreadPool(10);
         final ConnectionFactory factory = TestUtils.connectionFactory();
+        factory.setSocketConfigurator(new DefaultSocketConfigurator() {
+
+            int DEFAULT_RECEIVE_BUFFER_SIZE = 43690;
+
+            @Override
+            public void configure(Socket socket) throws IOException {
+                super.configure(socket);
+                socket.setReceiveBufferSize(DEFAULT_RECEIVE_BUFFER_SIZE);
+            }
+        });
         factory.setAutomaticRecoveryEnabled(true);
         factory.setTopologyRecoveryEnabled(true);
         // we try to set the lower values for closing timeouts, etc.
         // this makes the test execute faster.
-        factory.setShutdownExecutor(shutdownService);
-        factory.setShutdownTimeout(10000);
         factory.setRequestedHeartbeat(5);
-        factory.setSharedExecutor(dispatchingService);
-        factory.setNetworkRecoveryInterval(1000);
+        factory.setSharedExecutor(executorService);
+        // we need the shutdown executor: channel shutting down depends on the work pool,
+        // which is full. Channel shutting down will time out with the shutdown executor
+        factory.setShutdownExecutor(executorService);
+        factory.setNetworkRecoveryInterval(2000);
 
         producingConnection = (AutorecoveringConnection) factory.newConnection("Producer Connection");
         producingChannel = (AutorecoveringChannel) producingConnection.createChannel();
         consumingConnection = (AutorecoveringConnection) factory.newConnection("Consuming Connection");
         consumingChannel = (AutorecoveringChannel) consumingConnection.createChannel();
 
-        consumerRecoverOkLatch = new CountDownLatch(1);
+        consumerOkLatch = new CountDownLatch(2);
     }
 
     @After
     public void tearDown() throws IOException {
         closeConnectionIfOpen(consumingConnection);
         closeConnectionIfOpen(producingConnection);
 
-        dispatchingService.shutdownNow();
-        producerService.shutdownNow();
-        shutdownService.shutdownNow();
+        executorService.shutdownNow();
     }
 
     @Test
@@ -116,17 +121,17 @@ public void failureAndRecovery() throws IOException, InterruptedException {
         }
         final String queue = UUID.randomUUID().toString();
 
-        final CountDownLatch latch = new CountDownLatch(1);
+        final CountDownLatch recoveryLatch = new CountDownLatch(1);
 
         consumingConnection.addRecoveryListener(new RecoveryListener() {
 
             @Override
             public void handleRecovery(Recoverable recoverable) {
+                recoveryLatch.countDown();
             }
 
             @Override
             public void handleRecoveryStarted(Recoverable recoverable) {
-                latch.countDown();
             }
         });
 
@@ -136,12 +141,12 @@ public void handleRecoveryStarted(Recoverable recoverable) {
 
         assertThat(
             "Connection should have been closed and should have recovered by now",
-            latch.await(60, TimeUnit.SECONDS), is(true)
+            recoveryLatch.await(60, TimeUnit.SECONDS), is(true)
         );
 
         assertThat(
             "Consumer should have recovered by now",
-            latch.await(5, TimeUnit.SECONDS), is(true)
+            consumerOkLatch.await(5, TimeUnit.SECONDS), is(true)
         );
     }
 
@@ -158,12 +163,12 @@ private void declareQueue(final Channel channel, final String queue) throws IOEx
 
     private void produceMessagesInBackground(final Channel channel, final String queue) throws IOException {
         final AMQP.BasicProperties properties = new AMQP.BasicProperties.Builder().deliveryMode(1).build();
-        producerService.submit(new Callable<Void>() {
+        executorService.submit(new Callable<Void>() {
 
             @Override
             public Void call() throws Exception {
                 for (int i = 0; i < NUM_MESSAGES_TO_PRODUCE; i++) {
-                    channel.basicPublish("", queue, false, properties, ("MSG NUM" + i).getBytes());
+                    channel.basicPublish("", queue, false, properties, MESSAGE_CONTENT);
                 }
                 closeConnectionIfOpen(producingConnection);
                 return null;
@@ -172,36 +177,29 @@ public Void call() throws Exception {
     }
 
     private void startConsumer(final String queue) throws IOException {
-        consumingChannel.basicConsume(queue, false, "", false, false, null, new DefaultConsumer(consumingChannel) {
+        consumingChannel.basicConsume(queue, true, new DefaultConsumer(consumingChannel) {
 
             @Override
-            public void handleRecoverOk(String consumerTag) {
-                consumerRecoverOkLatch.countDown();
+            public void handleConsumeOk(String consumerTag) {
+                consumerOkLatch.countDown();
             }
 
             @Override
             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                 consumerWork();
                 try {
-                    consumingChannel.basicAck(envelope.getDeliveryTag(), false);
+                    consumingChannel.basicPublish("", "", null, "".getBytes());
                 } catch (Exception e) {
                     // application should handle writing exceptions
                 }
             }
         });
-        try {
-            Thread.sleep(500);
-        } catch (InterruptedException e) {
-            e.printStackTrace();
-        }
-        consumingChannel.basicQos(QOS_PREFETCH);
     }
 
     private void consumerWork() {
         try {
             Thread.sleep(MESSAGE_PROCESSING_TIME_MS);
         } catch (InterruptedException e) {
-            e.printStackTrace();
         }
     }
 }

src/test/resources/logback-test.xml

@@ -5,7 +5,7 @@
         </encoder>
     </appender>
 
-    <root level="info">
+    <root level="warn">
         <appender-ref ref="STDOUT" />
     </root>
 </configuration>