Do not use transient non-exclusive queue on RabbitMQ 4.3+

Fall back to durable non-exclusive.

References rabbitmq/rabbitmq-server#15459

Author: acogoluegnes

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

@@ -38,9 +38,13 @@
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class MulticastParams {
 
+  private static final Logger LOGGER = LoggerFactory.getLogger(MulticastParams.class);
+
   private long confirm = -1;
   private int confirmTimeout = 30;
   private int consumerCount = 1;
@@ -956,23 +960,24 @@ class State {
         TopologyRecording topologyRecording = state.topologyRecording;
 
         if (!params.predeclared || !queueExists(connection, qName)) {
+          boolean durable = params.flags.contains("persistent");
+          boolean exclusive = params.isExclusive();
+          boolean autoDelete = params.autoDelete;
+
+          if (Utils.atLeast4_3(connection) && !durable && !exclusive) {
+            LOGGER.warn(
+                "Non-durable non-exclusive queues are not supported on RabbitMQ 4.3+, "
+                    + "switching to durable non-exclusive");
+            durable = true;
+          }
+
           boolean serverNamed = qName == null || "".equals(qName);
           qName =
               channel
-                  .queueDeclare(
-                      qName,
-                      params.flags.contains("persistent"),
-                      params.isExclusive(),
-                      params.autoDelete,
-                      params.queueArguments)
+                  .queueDeclare(qName, durable, exclusive, autoDelete, params.queueArguments)
                   .getQueue();
           topologyRecording.recordQueue(
-              qName,
-              params.flags.contains("persistent"),
-              params.isExclusive(),
-              params.autoDelete,
-              params.queueArguments,
-              serverNamed);
+              qName, durable, exclusive, autoDelete, params.queueArguments, serverNamed);
         }
         generatedQueueNames.add(qName);
         // skipping binding to default exchange,

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

@@ -318,6 +318,33 @@ static boolean exists(Connection connection, Checker checker) throws IOException
     }
   }
 
+  static boolean atLeast4_3(Connection connection) {
+    String version = connection.getServerProperties().get("version").toString();
+    return versionCompare(version, "4.3.0") >= 0;
+  }
+
+  /**
+   * https://stackoverflow.com/questions/6701948/efficient-way-to-compare-version-strings-in-java
+   */
+  private static int versionCompare(String str1, String str2) {
+    String[] vals1 = str1.split("\\.");
+    String[] vals2 = str2.split("\\.");
+    int i = 0;
+    // set index to first non-equal ordinal or length of shortest version string
+    while (i < vals1.length && i < vals2.length && vals1[i].equals(vals2[i])) {
+      i++;
+    }
+    // compare first non-equal ordinal number
+    if (i < vals1.length && i < vals2.length) {
+      Integer val1 = Integer.valueOf(vals1[i]);
+      Integer val2 = Integer.valueOf(vals2[i]);
+      return val1.compareTo(val2);
+    }
+    // the strings are equal or one string is a substring of the other
+    // e.g. "1.2.3" = "1.2.3" or "1.2.3" < "1.2.3.4"
+    return Integer.signum(vals1.length - vals2.length);
+  }
+
   @SuppressWarnings("unchecked")
   static InstanceSynchronization defaultInstanceSynchronization(
       String id, int expectedInstances, String namespace, Duration timeout, PrintStream out) {

src/test/java/com/rabbitmq/perf/it/ConnectionRecoveryIT.java

@@ -381,7 +381,7 @@ public void shouldRecoverWithPreDeclared(boolean polling, TestInfo info) throws
     try (Connection c = connectionFactory.newConnection()) {
       Channel ch = c.createChannel();
       for (int i = 1; i <= queueCount; i++) {
-        ch.queueDeclare(String.format(queuePattern, i), false, false, false, null);
+        ch.queueDeclare(String.format(queuePattern, i), true, false, false, null);
       }
     }