Run SAC tests against RabbitMQ 3.11 or more

Refactor also the JUnit conditions to check the broker version.

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/impl/Utils.java

@@ -15,11 +15,11 @@
 
 import com.rabbitmq.stream.Address;
 import com.rabbitmq.stream.Constants;
+import com.rabbitmq.stream.ConsumerUpdateListener;
+import com.rabbitmq.stream.OffsetSpecification;
 import com.rabbitmq.stream.StreamDoesNotExistException;
 import com.rabbitmq.stream.StreamException;
 import com.rabbitmq.stream.StreamNotAvailableException;
-import com.rabbitmq.stream.ConsumerUpdateListener;
-import com.rabbitmq.stream.OffsetSpecification;
 import com.rabbitmq.stream.impl.Client.ClientParameters;
 import java.security.cert.X509Certificate;
 import java.time.Duration;

src/test/java/com/rabbitmq/stream/impl/ClientTest.java

@@ -41,6 +41,7 @@
 import com.rabbitmq.stream.impl.Client.StreamInfoResponse;
 import com.rabbitmq.stream.impl.Client.StreamParametersBuilder;
 import com.rabbitmq.stream.impl.ServerFrameHandler.FrameHandlerInfo;
+import com.rabbitmq.stream.impl.TestUtils.BrokerVersion;
 import com.rabbitmq.stream.impl.TestUtils.BrokerVersionAtLeast;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.UnpooledByteBufAllocator;
@@ -833,7 +834,7 @@ void closingPublisherWhilePublishingShouldNotCloseConnection(String publisherRef
   }
 
   @Test
-  @BrokerVersionAtLeast("3.11.0")
+  @BrokerVersionAtLeast(BrokerVersion.RABBITMQ_3_11)
   void exchangeCommandVersions() {
     Client client = cf.get();
     List<FrameHandlerInfo> infos = client.exchangeCommandVersions();
@@ -842,7 +843,7 @@ void exchangeCommandVersions() {
   }
 
   @Test
-  @BrokerVersionAtLeast("3.11.0")
+  @BrokerVersionAtLeast(BrokerVersion.RABBITMQ_3_11)
   void deliverVersion2LastCommittedOffsetShouldBeSet() throws Exception {
     int publishCount = 20_000;
     byte correlationId = 42;
@@ -881,7 +882,7 @@ void deliverVersion2LastCommittedOffsetShouldBeSet() throws Exception {
   }
 
   @Test
-  @BrokerVersionAtLeast("3.11.0")
+  @BrokerVersionAtLeast(BrokerVersion.RABBITMQ_3_11)
   void streamInfoShouldReturnFirstOffsetAndCommittedOffset() throws Exception {
     int publishCount = 20_000;
     CountDownLatch latch = new CountDownLatch(publishCount);
@@ -922,14 +923,14 @@ void streamInfoShouldReturnFirstOffsetAndCommittedOffset() throws Exception {
   }
 
   @Test
-  @BrokerVersionAtLeast("3.11.0")
+  @BrokerVersionAtLeast(BrokerVersion.RABBITMQ_3_11)
   void streamInfoShouldReturnErrorWhenStreamDoesNotExist() {
     assertThat(cf.get().streamStats("does not exist").getResponseCode())
         .isEqualTo(Constants.RESPONSE_CODE_STREAM_DOES_NOT_EXIST);
   }
 
   @Test
-  @BrokerVersionAtLeast("3.11.0")
+  @BrokerVersionAtLeast(BrokerVersion.RABBITMQ_3_11)
   void streamInfoFirstOffsetShouldChangeAfterRetentionKickedIn(TestInfo info) {
     int messageCount = 1000;
     int payloadSize = 1000;

src/test/java/com/rabbitmq/stream/impl/SacClientTest.java

@@ -35,6 +35,7 @@
 import com.rabbitmq.stream.impl.Client.CreditNotification;
 import com.rabbitmq.stream.impl.Client.MessageListener;
 import com.rabbitmq.stream.impl.Client.Response;
+import com.rabbitmq.stream.impl.TestUtils.BrokerVersionAtLeast311Condition;
 import com.rabbitmq.stream.impl.TestUtils.DisabledIfRabbitMqCtlNotSet;
 import java.nio.charset.StandardCharsets;
 import java.util.Collections;
@@ -53,7 +54,10 @@
 import org.junit.jupiter.api.TestInfo;
 import org.junit.jupiter.api.extension.ExtendWith;
 
-@ExtendWith(TestUtils.StreamTestInfrastructureExtension.class)
+@ExtendWith({
+  TestUtils.StreamTestInfrastructureExtension.class,
+  BrokerVersionAtLeast311Condition.class
+})
 @TestUtils.SingleActiveConsumer
 public class SacClientTest {
 

src/test/java/com/rabbitmq/stream/impl/SacStreamConsumerTest.java

@@ -22,6 +22,7 @@
 import com.rabbitmq.stream.Environment;
 import com.rabbitmq.stream.EnvironmentBuilder;
 import com.rabbitmq.stream.OffsetSpecification;
+import com.rabbitmq.stream.impl.TestUtils.BrokerVersionAtLeast311Condition;
 import io.netty.channel.EventLoopGroup;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -32,7 +33,10 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 
-@ExtendWith(TestUtils.StreamTestInfrastructureExtension.class)
+@ExtendWith({
+  TestUtils.StreamTestInfrastructureExtension.class,
+  BrokerVersionAtLeast311Condition.class
+})
 @TestUtils.SingleActiveConsumer
 public class SacStreamConsumerTest {
 

src/test/java/com/rabbitmq/stream/impl/SacSuperStreamConsumerTest.java

@@ -28,6 +28,7 @@
 import com.rabbitmq.stream.EnvironmentBuilder;
 import com.rabbitmq.stream.NoOffsetException;
 import com.rabbitmq.stream.OffsetSpecification;
+import com.rabbitmq.stream.impl.TestUtils.BrokerVersionAtLeast311Condition;
 import com.rabbitmq.stream.impl.TestUtils.CallableBooleanSupplier;
 import com.rabbitmq.stream.impl.TestUtils.SingleActiveConsumer;
 import com.rabbitmq.stream.impl.Utils.CompositeConsumerUpdateListener;
@@ -46,7 +47,10 @@
 import org.junit.jupiter.api.TestInfo;
 import org.junit.jupiter.api.extension.ExtendWith;
 
-@ExtendWith(TestUtils.StreamTestInfrastructureExtension.class)
+@ExtendWith({
+  TestUtils.StreamTestInfrastructureExtension.class,
+  BrokerVersionAtLeast311Condition.class
+})
 @SingleActiveConsumer
 public class SacSuperStreamConsumerTest {
   EventLoopGroup eventLoopGroup;

src/test/java/com/rabbitmq/stream/impl/StreamConsumerTest.java

@@ -36,6 +36,7 @@
 import com.rabbitmq.stream.StreamDoesNotExistException;
 import com.rabbitmq.stream.impl.Client.QueryOffsetResponse;
 import com.rabbitmq.stream.impl.MonitoringTestUtils.ConsumerInfo;
+import com.rabbitmq.stream.impl.TestUtils.BrokerVersion;
 import com.rabbitmq.stream.impl.TestUtils.BrokerVersionAtLeast;
 import com.rabbitmq.stream.impl.TestUtils.DisabledIfRabbitMqCtlNotSet;
 import io.netty.channel.EventLoopGroup;
@@ -141,7 +142,7 @@ void nameShouldBeSetIfTrackingStrategyIsSet() {
   }
 
   @Test
-  @BrokerVersionAtLeast("3.11.0")
+  @BrokerVersionAtLeast(BrokerVersion.RABBITMQ_3_11)
   void committedOffsetShouldBeSet() throws Exception {
     int messageCount = 20_000;
     TestUtils.publishAndWaitForConfirms(cf, messageCount, this.stream);

src/test/java/com/rabbitmq/stream/impl/StreamEnvironmentTest.java

@@ -45,6 +45,7 @@
 import com.rabbitmq.stream.StreamStats;
 import com.rabbitmq.stream.impl.Client.StreamMetadata;
 import com.rabbitmq.stream.impl.MonitoringTestUtils.EnvironmentInfo;
+import com.rabbitmq.stream.impl.TestUtils.BrokerVersion;
 import com.rabbitmq.stream.impl.TestUtils.BrokerVersionAtLeast;
 import com.rabbitmq.stream.impl.TestUtils.DisabledIfTlsNotEnabled;
 import io.netty.channel.EventLoopGroup;
@@ -504,7 +505,7 @@ void createPublishConsumeDelete(boolean lazyInit, TestInfo info) {
   }
 
   @Test
-  @BrokerVersionAtLeast("3.11.0")
+  @BrokerVersionAtLeast(BrokerVersion.RABBITMQ_3_11)
   void queryStreamInfoShouldReturnFirstOffsetAndCommittedOffset() throws Exception {
     try (Environment env = environmentBuilder.build()) {
       StreamStats info = env.queryStreamInfo(stream);
@@ -536,7 +537,7 @@ void queryStreamInfoShouldReturnFirstOffsetAndCommittedOffset() throws Exception
   }
 
   @Test
-  @BrokerVersionAtLeast("3.11.0")
+  @BrokerVersionAtLeast(BrokerVersion.RABBITMQ_3_11)
   void queryStreamInfoShouldThrowExceptionWhenStreamDoesNotExist() {
     try (Environment env = environmentBuilder.build()) {
       assertThatThrownBy(() -> env.queryStreamInfo("does not exist"))

src/test/java/com/rabbitmq/stream/impl/TestUtils.java

@@ -44,7 +44,6 @@
 import java.lang.annotation.Retention;
 import java.lang.annotation.RetentionPolicy;
 import java.lang.annotation.Target;
-import java.lang.reflect.AnnotatedElement;
 import java.lang.reflect.Field;
 import java.lang.reflect.Method;
 import java.nio.charset.StandardCharsets;
@@ -54,7 +53,6 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.ConcurrentHashMap;
@@ -468,10 +466,10 @@ static boolean atLeastVersion(String expectedVersion, String currentVersion) {
   @Target({ElementType.TYPE, ElementType.METHOD})
   @Retention(RetentionPolicy.RUNTIME)
   @Documented
-  @ExtendWith(BrokerVersionAtLeastCondition.class)
+  @ExtendWith(AnnotationBrokerVersionAtLeastCondition.class)
   public @interface BrokerVersionAtLeast {
 
-    String value();
+    BrokerVersion value();
   }
 
   interface TaskWithException {
@@ -746,29 +744,47 @@ public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext con
     }
   }
 
-  static class BrokerVersionAtLeastCondition implements ExecutionCondition {
+  private static class BaseBrokerVersionAtLeastCondition implements ExecutionCondition {
+
+    private final Function<ExtensionContext, String> versionProvider;
+
+    private BaseBrokerVersionAtLeastCondition(Function<ExtensionContext, String> versionProvider) {
+      this.versionProvider = versionProvider;
+    }
 
     @Override
     public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext context) {
-      Optional<AnnotatedElement> element = context.getElement();
-      //      String expectedVersion = annotation.get().getAnnotation(BrokerVersionAtLeast.class);
-      BrokerVersionAtLeast annotation = element.get().getAnnotation(BrokerVersionAtLeast.class);
-      if (annotation == null) {
+      if (!context.getTestMethod().isPresent()) {
+        return ConditionEvaluationResult.enabled("Apply only to methods");
+      }
+      String expectedVersion = versionProvider.apply(context);
+      if (expectedVersion == null) {
         return ConditionEvaluationResult.enabled("No broker version requirement");
       } else {
-        EventLoopGroup eventLoopGroup = StreamTestInfrastructureExtension.eventLoopGroup(context);
-        if (eventLoopGroup == null) {
-          throw new IllegalStateException(
-              "The event loop group must be in the test context to use "
-                  + BrokerVersionAtLeast.class.getSimpleName()
-                  + ", use the "
-                  + StreamTestInfrastructureExtension.class.getSimpleName()
-                  + " extension in the test");
-        }
-        Client client = new Client(new ClientParameters().eventLoopGroup(eventLoopGroup));
-        String expectedVersion = annotation.value();
-        String brokerVersion = client.brokerVersion();
-        client.close();
+        String brokerVersion =
+            context
+                .getRoot()
+                .getStore(Namespace.GLOBAL)
+                .getOrComputeIfAbsent(
+                    "brokerVersion",
+                    k -> {
+                      EventLoopGroup eventLoopGroup =
+                          StreamTestInfrastructureExtension.eventLoopGroup(context);
+                      if (eventLoopGroup == null) {
+                        throw new IllegalStateException(
+                            "The event loop group must be in the test context to use "
+                                + BrokerVersionAtLeast.class.getSimpleName()
+                                + ", use the "
+                                + StreamTestInfrastructureExtension.class.getSimpleName()
+                                + " extension in the test");
+                      }
+                      try (Client client =
+                          new Client(new ClientParameters().eventLoopGroup(eventLoopGroup))) {
+                        return client.brokerVersion();
+                      }
+                    },
+                    String.class);
+
         if (atLeastVersion(expectedVersion, brokerVersion)) {
           return ConditionEvaluationResult.enabled(
               "Broker version requirement met, expected "
@@ -786,6 +802,26 @@ public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext con
     }
   }
 
+  private static class AnnotationBrokerVersionAtLeastCondition
+      extends BaseBrokerVersionAtLeastCondition {
+
+    private AnnotationBrokerVersionAtLeastCondition() {
+      super(
+          context -> {
+            BrokerVersionAtLeast annotation =
+                context.getElement().get().getAnnotation(BrokerVersionAtLeast.class);
+            return annotation == null ? null : annotation.value().toString();
+          });
+    }
+  }
+
+  static class BrokerVersionAtLeast311Condition extends BaseBrokerVersionAtLeastCondition {
+
+    private BrokerVersionAtLeast311Condition() {
+      super(context -> "3.11.0");
+    }
+  }
+
   static class CountDownLatchAssert implements AssertDelegateTarget {
 
     private static final Duration TIMEOUT = Duration.ofSeconds(10);
@@ -851,4 +887,19 @@ static void waitMs(long waitTime) {
   @Retention(RetentionPolicy.RUNTIME)
   @Tag("single-active-consumer")
   public @interface SingleActiveConsumer {}
+
+  public enum BrokerVersion {
+    RABBITMQ_3_11("3.11.0");
+
+    final String value;
+
+    BrokerVersion(String value) {
+      this.value = value;
+    }
+
+    @Override
+    public String toString() {
+      return this.value;
+    }
+  }
 }