From de07734d766dc9822ff20ab7b69ea3081cb8f864 Mon Sep 17 00:00:00 2001
From: Mateusz Szczygiel <mateusz.szczygiel@allegro.pl>
Date: Thu, 25 Jan 2024 11:57:48 +0100
Subject: [PATCH] Minor tests stability improvements

---
 .../consumer/ConsumerMessageSenderTest.java   | 20 +++++++++----------
 .../setup/HermesExtension.java                |  4 ----
 .../PublishingAndConsumingTest.java           |  9 +++++++++
 3 files changed, 19 insertions(+), 14 deletions(-)

diff --git a/hermes-consumers/src/test/java/pl/allegro/tech/hermes/consumers/consumer/ConsumerMessageSenderTest.java b/hermes-consumers/src/test/java/pl/allegro/tech/hermes/consumers/consumer/ConsumerMessageSenderTest.java
index 5c1818aacc..42c1037b3b 100644
--- a/hermes-consumers/src/test/java/pl/allegro/tech/hermes/consumers/consumer/ConsumerMessageSenderTest.java
+++ b/hermes-consumers/src/test/java/pl/allegro/tech/hermes/consumers/consumer/ConsumerMessageSenderTest.java
@@ -25,7 +25,7 @@
 
 import java.nio.charset.StandardCharsets;
 import java.time.Clock;
-import java.util.Arrays;
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executors;
 
@@ -47,9 +47,9 @@
 public class ConsumerMessageSenderTest {
 
     public static final int ASYNC_TIMEOUT_MS = 2000;
-    private Subscription subscription = subscriptionWithTtl(10);
+    private final Subscription subscription = subscriptionWithTtl(10);
 
-    private Subscription subscriptionWith4xxRetry = subscriptionWithTtlAndClientErrorRetry(10);
+    private final Subscription subscriptionWith4xxRetry = subscriptionWithTtlAndClientErrorRetry(10);
 
     @Mock
     private MessageSender messageSender;
@@ -122,7 +122,7 @@ public void shouldHandleSuccessfulSending() {
     }
 
     @Test
-    public void shouldKeepTryingToSendMessageFailedSending() throws InterruptedException {
+    public void shouldKeepTryingToSendMessageFailedSending() {
         // given
         Message message = message();
         doReturn(failure()).doReturn(failure()).doReturn(success()).when(messageSender).send(message);
@@ -154,7 +154,7 @@ public void shouldDiscardMessageWhenTTLIsExceeded() {
     }
 
     @Test
-    public void shouldNotKeepTryingToSendMessageFailedWithStatusCode4xx() throws InterruptedException {
+    public void shouldNotKeepTryingToSendMessageFailedWithStatusCode4xx() {
         // given
         Message message = message();
         doReturn(failure(403)).doReturn(success()).when(messageSender).send(message);
@@ -170,7 +170,7 @@ public void shouldNotKeepTryingToSendMessageFailedWithStatusCode4xx() throws Int
     }
 
     @Test
-    public void shouldKeepTryingToSendMessageFailedWithStatusCode4xxForSubscriptionWith4xxRetry() throws InterruptedException {
+    public void shouldKeepTryingToSendMessageFailedWithStatusCode4xxForSubscriptionWith4xxRetry() {
         // given
         ConsumerMessageSender sender = consumerMessageSender(subscriptionWith4xxRetry);
         Message message = message();
@@ -227,11 +227,11 @@ public void shouldBackoffRetriesWhenEndpointFails() throws InterruptedException
     }
 
     @Test
-    public void shouldNotRetryOnRetryAfterAboveTtl() throws InterruptedException {
+    public void shouldNotRetryOnRetryAfterAboveTtl() {
         // given
         int retrySeconds = subscription.getSerialSubscriptionPolicy().getMessageTtl();
         Message message = message();
-        doReturn(backoff(retrySeconds)).when(messageSender).send(message);
+        doReturn(backoff(retrySeconds + 1)).when(messageSender).send(message);
 
         // when
         sender.sendAsync(message);
@@ -390,8 +390,8 @@ private ConsumerMessageSender consumerMessageSender(Subscription subscription) {
         ConsumerMessageSender sender = new ConsumerMessageSender(
                 subscription,
                 messageSenderFactory,
-                Arrays.asList(successHandler),
-                Arrays.asList(errorHandler),
+                List.of(successHandler),
+                List.of(errorHandler),
                 rateLimiter,
                 Executors.newSingleThreadExecutor(),
                 () -> inflightSemaphore.release(),
diff --git a/integration-tests/src/common/java/pl/allegro/tech/hermes/integrationtests/setup/HermesExtension.java b/integration-tests/src/common/java/pl/allegro/tech/hermes/integrationtests/setup/HermesExtension.java
index 45b1a0000b..fa0cac6b9e 100644
--- a/integration-tests/src/common/java/pl/allegro/tech/hermes/integrationtests/setup/HermesExtension.java
+++ b/integration-tests/src/common/java/pl/allegro/tech/hermes/integrationtests/setup/HermesExtension.java
@@ -139,10 +139,6 @@ private void removeGroups() {
                 logger.warn("Error during removing group: {}", group, e);
             }
         }
-
-        waitAtMost(adjust(Duration.ONE_MINUTE)).until(() ->
-                Assertions.assertThat(service.listGroups().size()).isEqualTo(0)
-        );
     }
 
     public void clearManagementData() {
diff --git a/integration-tests/src/integrationTest/java/pl/allegro/tech/hermes/integrationtests/PublishingAndConsumingTest.java b/integration-tests/src/integrationTest/java/pl/allegro/tech/hermes/integrationtests/PublishingAndConsumingTest.java
index 940cded8c9..79005283eb 100644
--- a/integration-tests/src/integrationTest/java/pl/allegro/tech/hermes/integrationtests/PublishingAndConsumingTest.java
+++ b/integration-tests/src/integrationTest/java/pl/allegro/tech/hermes/integrationtests/PublishingAndConsumingTest.java
@@ -5,6 +5,7 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 import org.springframework.test.web.reactive.server.WebTestClient;
+import pl.allegro.tech.hermes.api.ConsumerGroup;
 import pl.allegro.tech.hermes.api.ContentType;
 import pl.allegro.tech.hermes.api.Subscription;
 import pl.allegro.tech.hermes.api.SubscriptionMetrics;
@@ -25,6 +26,7 @@
 import java.time.Duration;
 import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.TimeUnit;
 
 import static com.jayway.awaitility.Awaitility.waitAtMost;
 import static com.jayway.awaitility.Duration.TEN_SECONDS;
@@ -222,6 +224,13 @@ public void shouldNotConsumeMessagesWhenSubscriptionIsSuspended() {
         hermes.api().waitUntilSubscriptionActivated(topic.getQualifiedName(), subscriptionName);
         hermes.api().suspendSubscription(topic, subscriptionName);
         hermes.api().waitUntilSubscriptionSuspended(topic.getQualifiedName(), subscriptionName);
+        // wait until consumer group removed
+        waitAtMost(30, TimeUnit.SECONDS).until(() ->
+                        hermes.api().getConsumerGroupsDescription(topic.getQualifiedName(), subscriptionName).expectBodyList(ConsumerGroup.class).returnResult().getResponseBody()
+                                .get(0)
+                                .getState()
+                                .equals("Empty")
+                );
         hermes.api().publishUntilSuccess(topic.getQualifiedName(),  TestMessage.of("hello", "world").body());
 
         // then