Exchange command versions if possible

For consumer connections, on RabbitMQ 3.11+. This is to benefit
from deliver v2 if possible.

References #158

Author: acogoluegnes

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

@@ -290,8 +290,8 @@ private MessageHandlerContext(
         long offset, long timestamp, long committedOffset, Consumer consumer) {
       this.offset = offset;
       this.timestamp = timestamp;
-      this.consumer = consumer;
       this.committedOffset = committedOffset;
+      this.consumer = consumer;
     }
 
     @Override
@@ -438,7 +438,7 @@ private ClientSubscriptionsManager(
                   subscriptionId & 0xFF,
                   Utils.formatConstant(responseCode));
       MessageListener messageListener =
-          (subscriptionId, offset, committedOffset, chunkTimestamp, message) -> {
+          (subscriptionId, offset, chunkTimestamp, committedOffset, message) -> {
             SubscriptionTracker subscriptionTracker =
                 subscriptionTrackers.get(subscriptionId & 0xFF);
             if (subscriptionTracker != null) {
@@ -559,6 +559,7 @@ private ClientSubscriptionsManager(
                       .metadataListener(metadataListener))
               .key(owner.name);
       this.client = clientFactory.client(clientFactoryContext);
+      maybeExchangeCommandVersions(client);
       clientInitializedInManager.set(true);
     }
 
@@ -848,4 +849,14 @@ public String toString() {
       return "SubscriptionContext{" + "offsetSpecification=" + offsetSpecification + '}';
     }
   }
+
+  private static void maybeExchangeCommandVersions(Client client) {
+    try {
+      if (Utils.is3_11_OrMore(client.brokerVersion())) {
+        client.exchangeCommandVersions();
+      }
+    } catch (Exception e) {
+      LOGGER.info("Error while exchanging command versions: {}", e.getMessage());
+    }
+  }
 }

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

@@ -240,4 +240,45 @@ static Function<ClientConnectionType, String> defaultConnectionNamingStrategy(St
   static boolean offsetBefore(long x, long y) {
     return Long.compareUnsigned(x, y) < 0;
   }
+
+  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;
+  }
+
+  /**
+   * 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 boolean is3_11_OrMore(String brokerVersion) {
+    return versionCompare(currentVersion(brokerVersion), "3.11.0") >= 0;
+  }
 }

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

@@ -37,6 +37,7 @@
 import com.rabbitmq.stream.impl.Client.Response;
 import com.rabbitmq.stream.impl.Client.StreamParametersBuilder;
 import com.rabbitmq.stream.impl.ServerFrameHandler.FrameHandlerInfo;
+import com.rabbitmq.stream.impl.TestUtils.BrokerVersionAtLeast;
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.UnpooledByteBufAllocator;
 import java.io.ByteArrayOutputStream;
@@ -830,7 +831,7 @@ void closingPublisherWhilePublishingShouldNotCloseConnection(String publisherRef
   }
 
   @Test
-  @Disabled
+  @BrokerVersionAtLeast("3.11.0")
   void exchangeCommandVersions() {
     Client client = cf.get();
     List<FrameHandlerInfo> infos = client.exchangeCommandVersions();

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

@@ -35,6 +35,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.BrokerVersionAtLeast;
 import com.rabbitmq.stream.impl.TestUtils.DisabledIfRabbitMqCtlNotSet;
 import io.netty.channel.EventLoopGroup;
 import java.time.Duration;
@@ -146,6 +147,30 @@ void nameShouldBeSetIfTrackingStrategyIsSet() {
         });
   }
 
+  @Test
+  @BrokerVersionAtLeast("3.11.0")
+  void committedOffsetShouldBeSet() throws Exception {
+    int messageCount = 20_000;
+    TestUtils.publishAndWaitForConfirms(cf, messageCount, this.stream);
+
+    CountDownLatch consumeLatch = new CountDownLatch(messageCount);
+    AtomicLong committedOffset = new AtomicLong();
+    Consumer consumer =
+        environment.consumerBuilder().stream(stream)
+            .offset(OffsetSpecification.first())
+            .messageHandler(
+                (context, message) -> {
+                  committedOffset.set(context.committedOffset());
+                  consumeLatch.countDown();
+                })
+            .build();
+
+    assertThat(consumeLatch.await(10, TimeUnit.SECONDS)).isTrue();
+    assertThat(committedOffset.get()).isNotZero();
+
+    consumer.close();
+  }
+
   @Test
   void consume() throws Exception {
     int messageCount = 100_000;

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

@@ -414,34 +414,15 @@ static boolean atLeastVersion(String expectedVersion, String currentVersion) {
     }
     try {
       currentVersion = currentVersion(currentVersion);
-      return "0.0.0".equals(currentVersion) || versionCompare(currentVersion, expectedVersion) >= 0;
+      return "0.0.0".equals(currentVersion) || Utils.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);
-  }
+
 
   @Target({ElementType.TYPE, ElementType.METHOD})
   @Retention(RetentionPolicy.RUNTIME)

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

@@ -37,6 +37,8 @@
 import java.util.function.Predicate;
 import java.util.stream.IntStream;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
 
 public class UtilsTest {
 
@@ -102,4 +104,24 @@ void testOffsetBefore() {
     assertThat(offsetBefore(Long.MAX_VALUE + 10, Long.MAX_VALUE + 10)).isFalse();
     assertThat(offsetBefore(Long.MAX_VALUE + 10, Long.MAX_VALUE + 20)).isTrue();
   }
+
+  @ParameterizedTest
+  @CsvSource({
+    "3.8.0+rc.1.2186.g95f3fde,false",
+    "3.9.21,false",
+    "3.9.22-alpha.13,false",
+    "3.10.6,false",
+    "3.11.0-alpha.15,true",
+    "3.11.0,true",
+    "3.11.1,true",
+    "4.0.0-alpha.15,true",
+    "4.0.0,true",
+    "4.0.1,true",
+    "4.1.0-alpha.15,true",
+    "4.1.0,true",
+    "4.1.1,true",
+  })
+  void is_3_11_OrMore(String input, boolean expected) {
+    assertThat(Utils.is3_11_OrMore(input)).isEqualTo(expected);
+  }
 }