Add "at least broker version" condition for partition tests

They run only on latest alphas.

Author: acogoluegnes

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

@@ -1320,6 +1320,10 @@ int serverAdvertisedPort() {
     return Integer.valueOf(this.connectionProperties("advertised_port"));
   }
 
+  String brokerVersion() {
+    return this.serverProperties.get("version");
+  }
+
   private String connectionProperties(String key) {
     if (this.connectionProperties != null && this.connectionProperties.containsKey(key)) {
       return this.connectionProperties.get(key);

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

@@ -20,6 +20,7 @@
 
 import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
+import com.rabbitmq.stream.impl.TestUtils.BrokerVersionAtLeast;
 import java.util.List;
 import java.util.UUID;
 import java.util.stream.IntStream;
@@ -52,6 +53,7 @@ void tearDown() throws Exception {
   }
 
   @Test
+  @BrokerVersionAtLeast("3.9.6")
   void routeShouldReturnEmptyListWhenExchangeDoesNotExist() {
     assertThat(cf.get().route("", UUID.randomUUID().toString())).isEmpty();
   }
@@ -62,6 +64,7 @@ void partitionsShouldReturnEmptyListWhenExchangeDoesNotExist() {
   }
 
   @Test
+  @BrokerVersionAtLeast("3.9.6")
   void routeShouldReturnNullWhenNoStreamForRoutingKey() throws Exception {
     declareSuperStreamTopology(connection, superStream, partitions);
 
@@ -79,6 +82,7 @@ void partitionsShouldReturnEmptyListWhenThereIsNoBinding() throws Exception {
   }
 
   @Test
+  @BrokerVersionAtLeast("3.9.6")
   void routeTopologyWithPartitionCount() throws Exception {
     declareSuperStreamTopology(connection, superStream, 3);
 
@@ -94,6 +98,7 @@ void routeTopologyWithPartitionCount() throws Exception {
   }
 
   @Test
+  @BrokerVersionAtLeast("3.9.6")
   void routeReturnsMultipleStreamsIfMultipleBindingsForSameKey() throws Exception {
     declareSuperStreamTopology(connection, superStream, 3);
     connection.createChannel().queueBind(superStream + "-1", superStream, "0");

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

@@ -27,6 +27,7 @@
 import com.rabbitmq.stream.OffsetSpecification;
 import com.rabbitmq.stream.Producer;
 import com.rabbitmq.stream.ProducerBuilder.RoutingType;
+import com.rabbitmq.stream.impl.TestUtils.BrokerVersionAtLeast;
 import io.netty.channel.EventLoopGroup;
 import java.util.Map;
 import java.util.UUID;
@@ -74,6 +75,7 @@ void tearDown() throws Exception {
   }
 
   @Test
+  @BrokerVersionAtLeast("3.9.6")
   void allMessagesSentToSuperStreamWithHashRoutingShouldBeThenConsumed() throws Exception {
     int messageCount = 10_000;
     declareSuperStreamTopology(connection, superStream, partitions);
@@ -125,6 +127,7 @@ void allMessagesSentToSuperStreamWithHashRoutingShouldBeThenConsumed() throws Ex
   }
 
   @Test
+  @BrokerVersionAtLeast("3.9.6")
   void allMessagesSentToSuperStreamWithRoutingKeyRoutingShouldBeThenConsumed() throws Exception {
     int messageCount = 10_000;
     routingKeys = new String[] {"amer", "emea", "apac"};
@@ -176,6 +179,7 @@ void allMessagesSentToSuperStreamWithRoutingKeyRoutingShouldBeThenConsumed() thr
   }
 
   @Test
+  @BrokerVersionAtLeast("3.9.6")
   void getLastPublishingIdShouldReturnLowestValue() throws Exception {
     int messageCount = 10_000;
     declareSuperStreamTopology(connection, superStream, partitions);

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

@@ -27,6 +27,7 @@
 import com.rabbitmq.stream.MessageBuilder;
 import com.rabbitmq.stream.StreamException;
 import com.rabbitmq.stream.impl.Client.Broker;
+import com.rabbitmq.stream.impl.Client.ClientParameters;
 import com.rabbitmq.stream.impl.Client.StreamMetadata;
 import io.netty.channel.EventLoopGroup;
 import io.netty.channel.nio.NioEventLoopGroup;
@@ -39,6 +40,7 @@
 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;
@@ -47,6 +49,7 @@
 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;
@@ -71,6 +74,7 @@
 import org.junit.jupiter.api.extension.ExtensionContext;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
+import org.slf4j.LoggerFactory;
 
 public final class TestUtils {
 
@@ -363,6 +367,15 @@ static Map<String, StreamMetadata> metadata(Broker leader, List<Broker> replicas
   @ExtendWith(DisabledIfTlsNotEnabledCondition.class)
   public @interface DisabledIfTlsNotEnabled {}
 
+  @Target({ElementType.TYPE, ElementType.METHOD})
+  @Retention(RetentionPolicy.RUNTIME)
+  @Documented
+  @ExtendWith(BrokerVersionAtLeastCondition.class)
+  public @interface BrokerVersionAtLeast {
+
+    String value();
+  }
+
   interface TaskWithException {
 
     void run(Object context) throws Exception;
@@ -398,7 +411,7 @@ private static ExtensionContext.Store store(ExtensionContext extensionContext) {
       return extensionContext.getRoot().getStore(NAMESPACE);
     }
 
-    private static EventLoopGroup eventLoopGroup(ExtensionContext context) {
+    static EventLoopGroup eventLoopGroup(ExtensionContext context) {
       return (EventLoopGroup) store(context).get("nettyEventLoopGroup");
     }
 
@@ -570,6 +583,96 @@ public ConditionEvaluationResult evaluateExecutionCondition(ExtensionContext con
     }
   }
 
+  static class BrokerVersionAtLeastCondition implements ExecutionCondition {
+
+    @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) {
+        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();
+        if (atLeastVersion(expectedVersion, brokerVersion)) {
+          return ConditionEvaluationResult.enabled(
+              "Broker version requirement met, expected "
+                  + expectedVersion
+                  + ", actual "
+                  + brokerVersion);
+        } else {
+          return ConditionEvaluationResult.disabled(
+              "Broker version requirement not met, expected "
+                  + expectedVersion
+                  + ", actual "
+                  + brokerVersion);
+        }
+      }
+    }
+  }
+
+  private static String currentVersion(String currentVersion) {
+    // versions built from source: 3.7.0+rc.1.4.gedc5d96
+    if (currentVersion.contains("+")) {
+      currentVersion = currentVersion.substring(0, currentVersion.indexOf("+"));
+    }
+    // alpha (snapshot) versions: 3.7.0~alpha.449-1
+    if (currentVersion.contains("~")) {
+      currentVersion = currentVersion.substring(0, currentVersion.indexOf("~"));
+    }
+    // alpha (snapshot) versions: 3.7.1-alpha.40
+    if (currentVersion.contains("-")) {
+      currentVersion = currentVersion.substring(0, currentVersion.indexOf("-"));
+    }
+    return currentVersion;
+  }
+
+  static boolean atLeastVersion(String expectedVersion, String currentVersion) {
+    if (currentVersion.contains("alpha-stream")) {
+      return true;
+    }
+    try {
+      currentVersion = currentVersion(currentVersion);
+      return "0.0.0".equals(currentVersion) || versionCompare(currentVersion, expectedVersion) >= 0;
+    } catch (RuntimeException e) {
+      LoggerFactory.getLogger(TestUtils.class)
+          .warn("Unable to parse broker version {}", currentVersion, e);
+      throw e;
+    }
+  }
+
+  /**
+   * https://stackoverflow.com/questions/6701948/efficient-way-to-compare-version-strings-in-java
+   */
+  static int versionCompare(String str1, String str2) {
+    String[] vals1 = str1.split("\\.");
+    String[] vals2 = str2.split("\\.");
+    int i = 0;
+    // set index to first non-equal ordinal or length of shortest version string
+    while (i < vals1.length && i < vals2.length && vals1[i].equals(vals2[i])) {
+      i++;
+    }
+    // compare first non-equal ordinal number
+    if (i < vals1.length && i < vals2.length) {
+      int diff = Integer.valueOf(vals1[i]).compareTo(Integer.valueOf(vals2[i]));
+      return Integer.signum(diff);
+    }
+    // the strings are equal or one string is a substring of the other
+    // e.g. "1.2.3" = "1.2.3" or "1.2.3" < "1.2.3.4"
+    return Integer.signum(vals1.length - vals2.length);
+  }
+
   static class CountDownLatchAssert implements AssertDelegateTarget {
 
     private static final Duration TIMEOUT = Duration.ofSeconds(10);

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

@@ -0,0 +1,32 @@
+// Copyright (c) 2021 VMware, Inc. or its affiliates.  All rights reserved.
+//
+// This software, the RabbitMQ Stream Java client library, is dual-licensed under the
+// Mozilla Public License 2.0 ("MPL"), and the Apache License version 2 ("ASL").
+// For the MPL, please see LICENSE-MPL-RabbitMQ. For the ASL,
+// please see LICENSE-APACHE2.
+//
+// This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY KIND,
+// either express or implied. See the LICENSE file for specific language governing
+// rights and limitations of this software.
+//
+// If you have any questions regarding licensing, please contact us at
+// info@rabbitmq.com.
+package com.rabbitmq.stream.impl;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+public class TestUtilsTest {
+
+  @ParameterizedTest
+  @CsvSource({
+    "3.9.6,3.9.5,false",
+    "3.9.6,3.9.0-alpha-stream.232,true",
+    "3.9.6,3.9.6-alpha.28,true"
+  })
+  void atLeastVersion(String expectedVersion, String currentVersion, boolean expected) {
+    assertThat(TestUtils.atLeastVersion(expectedVersion, currentVersion)).isEqualTo(expected);
+  }
+}