Force ClientFutureFactory to "progressive"

Author: acogoluegnes

src/main/java/com/rabbitmq/client/amqp/impl/AmqpEnvironment.java

@@ -26,6 +26,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import org.apache.qpid.protonj2.client.*;
+import org.apache.qpid.protonj2.client.futures.ClientFutureFactory;
 
 class AmqpEnvironment implements Environment {
 
@@ -55,7 +56,7 @@ class AmqpEnvironment implements Environment {
     this.id = ID_SEQUENCE.getAndIncrement();
     connectionSettings.copyTo(this.connectionSettings);
     this.connectionSettings.consolidate();
-    ClientOptions clientOptions = new ClientOptions();
+    ClientOptions clientOptions = new ClientOptions().futureType(ClientFutureFactory.PROGRESSIVE);
     this.client = Client.create(clientOptions);
 
     if (executorService == null) {

src/test/java/com/rabbitmq/client/amqp/impl/RecoveryClusterTest.java

@@ -42,7 +42,6 @@
 import org.slf4j.LoggerFactory;
 
 @DisabledIfNotCluster
-@Disabled
 public class RecoveryClusterTest {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(RecoveryClusterTest.class);
@@ -60,20 +59,21 @@ public class RecoveryClusterTest {
   @BeforeAll
   static void initAll() {
     nodes = Cli.nodes();
+    LOGGER.info("Available processor(s): {}", Runtime.getRuntime().availableProcessors());
   }
 
   @BeforeEach
   void init(TestInfo info) {
     environment =
         new AmqpEnvironmentBuilder().connectionSettings().uris(URIS).environmentBuilder().build();
     this.connection =
-        (AmqpConnection)
-            environment
-                .connectionBuilder()
-                .recovery()
-                .backOffDelayPolicy(BACK_OFF_DELAY_POLICY)
-                .connectionBuilder()
-                .build();
+        connection(
+            b ->
+                b.name("c-management")
+                    .recovery()
+                    .backOffDelayPolicy(BACK_OFF_DELAY_POLICY)
+                    .connectionBuilder()
+                    .build());
     this.management = connection.management();
     this.testInfo = info;
   }
@@ -92,22 +92,36 @@ void clusterRestart() {
     List<ConsumerState> consumerStates = Collections.emptyList();
     try {
       qqNames.forEach(n -> management.queue(n).type(Management.QueueType.QUORUM).declare());
+      AtomicInteger counter = new AtomicInteger(0);
       consumerStates =
           qqNames.stream()
               .map(
                   n ->
                       new ConsumerState(
                           n,
-                          connection(b -> b.affinity().queue(n).operation(CONSUME).connection())))
+                          connection(
+                              b ->
+                                  b.name("consumer-" + counter.getAndIncrement())
+                                      .affinity()
+                                      .queue(n)
+                                      .operation(CONSUME)
+                                      .connection())))
               .collect(toList());
 
+      counter.set(0);
       publisherStates =
           qqNames.stream()
               .map(
                   n ->
                       new PublisherState(
                           n,
-                          connection(b -> b.affinity().queue(n).operation(PUBLISH).connection())))
+                          connection(
+                              b ->
+                                  b.name("publisher-" + counter.getAndIncrement())
+                                      .affinity()
+                                      .queue(n)
+                                      .operation(PUBLISH)
+                                      .connection())))
               .collect(toList());
 
       publisherStates.forEach(PublisherState::start);