Simplify test

Author: acogoluegnes

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

@@ -39,11 +39,18 @@
 import com.rabbitmq.client.amqp.impl.TestUtils.Sync;
 import java.util.List;
 import java.util.Set;
-import java.util.concurrent.*;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.IntStream;
 import org.assertj.core.api.Assertions;
-import org.junit.jupiter.api.*;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.ValueSource;
 import org.slf4j.Logger;
@@ -649,21 +656,6 @@ void shouldRecoverEvenIfManagementIsClosed(TestInfo info) {
       Management management = connection.management();
       Management.QueueInfo queueInfo = management.queue().exclusive(true).declare();
       String q = queueInfo.name();
-      Publisher publisher = connection.publisherBuilder().queue(q).build();
-      Sync publishSync = TestUtils.sync();
-      Publisher.Callback callback =
-          ctx -> {
-            if (ctx.status() == ACCEPTED) {
-              publishSync.down();
-            } else {
-              LOGGER.warn(
-                  "Unexpected status: {} ({})", ctx.status(), info.getTestMethod().get().getName());
-            }
-          };
-      publisher.publish(publisher.message(), callback);
-      assertThat(publishSync).completes();
-
-      waitAtMost(() -> management.queueInfo(q).messageCount() == 1);
 
       Sync consumeSync = TestUtils.sync();
       connection
@@ -676,16 +668,28 @@ void shouldRecoverEvenIfManagementIsClosed(TestInfo info) {
               })
           .build();
 
-      assertThat(consumeSync).completes();
-      waitAtMost(() -> management.queueInfo(q).messageCount() == 0);
-      management.close();
+      waitAtMost(() -> management.queueInfo(q).consumerCount() == 1);
 
-      publishSync.reset();
-      consumeSync.reset();
+      management.close();
 
       closeConnectionAndWaitForRecovery();
+
+      Publisher publisher = connection.publisherBuilder().queue(q).build();
+      Sync publishSync = TestUtils.sync();
+      Publisher.Callback callback =
+          ctx -> {
+            if (ctx.status() == ACCEPTED) {
+              publishSync.down();
+            } else {
+              LOGGER.warn(
+                  "Unexpected status: {} ({})", ctx.status(), info.getTestMethod().get().getName());
+            }
+          };
       publisher.publish(publisher.message(), callback);
+      assertThat(publishSync).completes();
+
       assertThat(consumeSync).completes();
+
       assertThatThrownBy(() -> management.queueInfo(q))
           .isInstanceOf(AmqpResourceClosedException.class);
       assertThat(connection.management().queueInfo(q)).isEmpty();