Handle connection and topology recovery

Use Java client connection recovery but uses specific topology recovery.
Let each connection (configuration, producer, consumer) recovers the
resources it needs.

[#159130745]

Fixes #106

Author: acogoluegnes

src/main/java/com/rabbitmq/perf/AgentBase.java

@@ -17,16 +17,19 @@
 
 import com.rabbitmq.client.MissedHeartbeatException;
 import com.rabbitmq.client.ShutdownSignalException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
-import java.util.function.BooleanSupplier;
 import java.util.function.Predicate;
 
 /**
  *
  */
 public abstract class AgentBase {
 
+    private static final Logger LOGGER = LoggerFactory.getLogger(AgentBase.class);
+
     // FIXME this is the condition to start connection recovery
     // ensure it's the appropriate condition and get it from the Java client code
     static final Predicate<ShutdownSignalException> CONNECTION_RECOVERY_TRIGGERED =
@@ -54,25 +57,38 @@ protected boolean isConnectionRecoveryTriggered(ShutdownSignalException e) {
         return CONNECTION_RECOVERY_TRIGGERED.test(e);
     }
 
-    protected void handleShutdownSignalExceptionOnWrite(BooleanSupplier showStopper, ShutdownSignalException e) {
-        if (shouldStop(showStopper, e)) {
+    protected void handleShutdownSignalExceptionOnWrite(Recovery.RecoveryProcess recoveryProcess, ShutdownSignalException e) {
+        if (LOGGER.isDebugEnabled()) {
+            LOGGER.debug(
+                "Handling write error, recovery process enabled? {}, condition to trigger connection recovery? {}",
+                recoveryProcess.isEnabled(), isConnectionRecoveryTriggered(e)
+            );
+        }
+        if (shouldStop(recoveryProcess, e)) {
             throw e;
         }
     }
 
-    protected boolean shouldStop(BooleanSupplier showStopper, ShutdownSignalException e) {
-        return !showStopper.getAsBoolean() || !isConnectionRecoveryTriggered(e);
+    protected boolean shouldStop(Recovery.RecoveryProcess recoveryProcess, ShutdownSignalException e) {
+        if (recoveryProcess.isEnabled()) {
+            // we stop only if the error isn't likely to trigger connection recovery
+            return !isConnectionRecoveryTriggered(e);
+        } else {
+            return true;
+        }
     }
 
-    protected void dealWithWriteOperation(WriteOperation writeOperation, BooleanSupplier showStopper) throws IOException {
+    protected void dealWithWriteOperation(WriteOperation writeOperation, Recovery.RecoveryProcess recoveryProcess) throws IOException {
         try {
             writeOperation.call();
         } catch (ShutdownSignalException e) {
-            handleShutdownSignalExceptionOnWrite(showStopper, e);
+            handleShutdownSignalExceptionOnWrite(recoveryProcess, e);
             // connection recovery is in progress, we ignore the exception if this point is reached
         }
     }
 
+    public abstract void recover(TopologyRecording topologyRecording);
+
     protected interface AgentState {
 
         float getRateLimit();

src/main/java/com/rabbitmq/perf/Consumer.java

@@ -20,6 +20,8 @@
 import com.rabbitmq.client.DefaultConsumer;
 import com.rabbitmq.client.Envelope;
 import com.rabbitmq.client.ShutdownSignalException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import java.io.ByteArrayInputStream;
 import java.io.DataInputStream;
@@ -31,11 +33,12 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.function.BiFunction;
-import java.util.function.BooleanSupplier;
 
 public class Consumer extends AgentBase implements Runnable {
 
-    private ConsumerImpl                q;
+    private static final Logger LOGGER = LoggerFactory.getLogger(Consumer.class);
+
+    private volatile ConsumerImpl       q;
     private final Channel               channel;
     private final String                id;
     private final List<String>          queueNames;
@@ -53,15 +56,15 @@ public class Consumer extends AgentBase implements Runnable {
 
     private final ConsumerState state;
 
-    private final BooleanSupplier showStopper;
+    private final Recovery.RecoveryProcess recoveryProcess;
 
     public Consumer(Channel channel, String id,
                     List<String> queueNames, int txSize, boolean autoAck,
                     int multiAckEvery, Stats stats, float rateLimit, int msgLimit,
                     int consumerLatencyInMicroSeconds,
                     TimestampProvider timestampProvider,
                     MulticastSet.CompletionHandler completionHandler,
-                    BooleanSupplier showStopper) {
+                    Recovery.RecoveryProcess recoveryProcess) {
 
         this.channel           = channel;
         this.id                = id;
@@ -101,7 +104,8 @@ public Consumer(Channel channel, String id,
         }
 
         this.state = new ConsumerState(rateLimit);
-        this.showStopper = showStopper;
+        this.recoveryProcess = recoveryProcess;
+        this.recoveryProcess.init(this);
     }
 
     public void run() {
@@ -140,11 +144,11 @@ public void handleDelivery(String consumerTag, Envelope envelope, BasicPropertie
                         } else if (currentMessageCount % multiAckEvery == 0) {
                             channel.basicAck(envelope.getDeliveryTag(), true);
                         }
-                    }, showStopper);
+                    }, recoveryProcess);
                 }
 
                 if (txSize != 0 && currentMessageCount % txSize == 0) {
-                    dealWithWriteOperation(() -> channel.txCommit(), showStopper);
+                    dealWithWriteOperation(() -> channel.txCommit(), recoveryProcess);
                 }
 
                 long diff_time = timestampProvider.getDifference(nowTimestamp, messageTimestamp);
@@ -163,7 +167,14 @@ public void handleDelivery(String consumerTag, Envelope envelope, BasicPropertie
 
         @Override
         public void handleShutdownSignal(String consumerTag, ShutdownSignalException sig) {
-            countDown();
+            LOGGER.debug(
+                "Consumer received shutdown signal, recovery process enabled? {}, condition to trigger connection recovery? {}",
+                recoveryProcess.isEnabled(), isConnectionRecoveryTriggered(sig)
+            );
+            if (!recoveryProcess.isEnabled()) {
+                LOGGER.debug("Counting down for consumer");
+                countDown();
+            }
         }
 
         @Override
@@ -185,6 +196,21 @@ private void countDown() {
         }
     }
 
+    @Override
+    public void recover(TopologyRecording topologyRecording) {
+        for (Map.Entry<String, String> entry : consumerTagBranchMap.entrySet()) {
+            TopologyRecording.RecordedQueue queue = topologyRecording.queue(entry.getValue());
+            try {
+                channel.basicConsume(queue.name(), autoAck, entry.getKey(), q);
+            } catch (IOException e) {
+                LOGGER.warn(
+                    "Error while recovering consumer {} on queue {} on connection {}",
+                    entry.getKey(), queue.name(), channel.getConnection().getClientProvidedName(), e
+                );
+            }
+        }
+    }
+
     private static class ConsumerState implements AgentState {
 
         private final float rateLimit;