Log unfinished multi-threaded topology recovery tasks

References #370

Author: acogoluegnes

src/main/java/com/rabbitmq/client/ConnectionFactory.java

@@ -719,6 +719,7 @@ public void setTopologyRecoveryEnabled(boolean topologyRecovery) {
     /**
      * Get the executor to use for parallel topology recovery. If null (the default), recovery is done single threaded on the main connection thread.
      * @return thread pool executor
+     * @since 4.7.0
      */
     public ExecutorService getTopologyRecoveryExecutor() {
         return topologyRecoveryExecutor;
@@ -727,8 +728,10 @@ public ExecutorService getTopologyRecoveryExecutor() {
     /**
      * Set the executor to use for parallel topology recovery. If null (the default), recovery is done single threaded on the main connection thread.
      * It is recommended to pass a ThreadPoolExecutor that will allow its core threads to timeout so these threads can die when recovery is complete.
+     * It's developer's responsibility to shut down the executor when it is no longer needed.
      * Note: your {@link ExceptionHandler#handleTopologyRecoveryException(Connection, Channel, TopologyRecoveryException)} method should be thread-safe.
      * @param topologyRecoveryExecutor thread pool executor
+     * @since 4.7.0
      */
     public void setTopologyRecoveryExecutor(final ExecutorService topologyRecoveryExecutor) {
         this.topologyRecoveryExecutor = topologyRecoveryExecutor;

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

@@ -32,7 +32,9 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
 import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
@@ -638,10 +640,10 @@ private void recoverTopology(final ExecutorService executor) {
             // We also need to recover 1 type of entity at a time in case channel1 has a binding to a queue that is currently owned and being recovered by channel2 for example
             // Note: invokeAll will block until all callables are completed and all returned futures will be complete 
             try {
-                executor.invokeAll(groupEntitiesByChannel(Utility.copy(recordedExchanges).values()));
-                executor.invokeAll(groupEntitiesByChannel(Utility.copy(recordedQueues).values()));
-                executor.invokeAll(groupEntitiesByChannel(Utility.copy(recordedBindings)));
-                executor.invokeAll(groupEntitiesByChannel(Utility.copy(consumers).values()));
+                recoverEntitiesAsynchronously(executor, Utility.copy(recordedExchanges).values());
+                recoverEntitiesAsynchronously(executor, Utility.copy(recordedQueues).values());
+                recoverEntitiesAsynchronously(executor, Utility.copy(recordedBindings));
+                recoverEntitiesAsynchronously(executor, Utility.copy(consumers).values());
             } catch (final Exception cause) {
                 final String message = "Caught an exception while recovering toplogy: " + cause.getMessage();
                 final TopologyRecoveryException e = new TopologyRecoveryException(message, cause);
@@ -748,7 +750,22 @@ private void propagateQueueNameChangeToConsumers(String oldName, String newName)
             }
         }
     }
-    
+
+    private void recoverEntitiesAsynchronously(ExecutorService executor, Collection<? extends RecordedEntity> recordedEntities) throws InterruptedException {
+        List<Future<Object>> tasks = executor.invokeAll(groupEntitiesByChannel(recordedEntities));
+        for (Future<Object> task : tasks) {
+            if (!task.isDone()) {
+                LOGGER.warn("Recovery task should be done {}", task);
+            } else {
+                try {
+                    task.get(1, TimeUnit.MILLISECONDS);
+                } catch (Exception e) {
+                    LOGGER.warn("Recovery task is done but returned an exception", e);
+                }
+            }
+        }
+    }
+
     private <E extends RecordedEntity> List<Callable<Object>> groupEntitiesByChannel(final Collection<E> entities) {
         // map entities by channel
         final Map<AutorecoveringChannel, List<E>> map = new LinkedHashMap<AutorecoveringChannel, List<E>>();