Add Javadoc for stream info command

References rabbitmq/rabbitmq-server#5412

Author: acogoluegnes

src/main/java/com/rabbitmq/stream/Environment.java

@@ -61,6 +61,15 @@ static EnvironmentBuilder builder() {
    */
   void deleteStream(String stream);
 
+  /**
+   * Query information on a stream.
+   *
+   * <p>Requires RabbitMQ 3.11 or more.
+   *
+   * @param stream
+   * @return information on the stream
+   * @throws UnsupportedOperationException if the broker does not support this command
+   */
   StreamInfo queryStreamInfo(String stream);
 
   /**

src/main/java/com/rabbitmq/stream/MessageHandler.java

@@ -54,7 +54,7 @@ interface Context {
     long timestamp();
 
     /**
-     * The committed offset on this stream.
+     * The committed offset in this stream.
      *
      * <p>It is the offset of the last message confirmed by a quorum of the stream cluster members
      * (leader and replicas).
@@ -63,7 +63,10 @@ interface Context {
      * given time. The value can be stale as soon as the application reads it though, as the
      * committed offset for a stream that is published to changes all the time.
      *
-     * @return committed offset on this stream
+     * <p>This requires RabbitMQ 3.11 or more. The method always returns 0 otherwise.
+     *
+     * @return committed offset in this stream
+     * @see StreamInfo#committedOffset()
      */
     long committedOffset();
 

src/main/java/com/rabbitmq/stream/StreamInfo.java

@@ -13,9 +13,36 @@
 // info@rabbitmq.com.
 package com.rabbitmq.stream;
 
+import com.rabbitmq.stream.MessageHandler.Context;
+
+/**
+ * Information on a stream.
+ *
+ * @see Environment#queryStreamInfo(String)
+ */
 public interface StreamInfo {
 
+  /**
+   * The first offset in the stream.
+   *
+   * @return first offset in the stream
+   * @throws NoOffsetException if there is no first offset yet
+   */
   long firstOffset();
 
+  /**
+   * The committed offset in the stream.
+   *
+   * <p>It is the offset of the last message confirmed by a quorum of the stream cluster members
+   * (leader and replicas).
+   *
+   * <p>The committed offset is a good indication of what the last offset of a stream is at a given
+   * time. The value can be stale as soon as the application reads it though, as the committed
+   * offset for a stream that is published to changes all the time.
+   *
+   * @return committed offset in this stream
+   * @see Context#committedOffset()
+   * @throws NoOffsetException if there is no committed offset yet
+   */
   long committedOffset();
 }

src/main/java/com/rabbitmq/stream/StreamNotAvailableException.java

@@ -13,12 +13,10 @@
 // info@rabbitmq.com.
 package com.rabbitmq.stream;
 
+/** Exception to indicate a stream is not available. */
 public class StreamNotAvailableException extends StreamException {
 
-  private final String stream;
-
   public StreamNotAvailableException(String stream) {
     super("Stream " + stream + " is not available", Constants.RESPONSE_CODE_STREAM_NOT_AVAILABLE);
-    this.stream = stream;
   }
 }

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

@@ -391,7 +391,16 @@ public void deleteStream(String stream) {
 
   @Override
   public StreamInfo queryStreamInfo(String stream) {
-    StreamInfoResponse response = locatorOperation(client -> client.streamInfo(stream));
+    StreamInfoResponse response =
+        locatorOperation(
+            client -> {
+              if (Utils.is3_11_OrMore(client.brokerVersion())) {
+                return client.streamInfo(stream);
+              } else {
+                throw new UnsupportedOperationException(
+                    "QueryStringInfo is available only for RabbitMQ 3.11 or more.");
+              }
+            });
     if (response.isOk()) {
       Map<String, String> info = response.getInfo();
       BiFunction<String, String, LongSupplier> offsetSupplierLogic =

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

@@ -922,12 +922,14 @@ void streamInfoShouldReturnFirstOffsetAndCommittedOffset() throws Exception {
   }
 
   @Test
+  @BrokerVersionAtLeast("3.11.0")
   void streamInfoShouldReturnErrorWhenStreamDoesNotExist() {
     assertThat(cf.get().streamInfo("does not exist").getResponseCode())
         .isEqualTo(Constants.RESPONSE_CODE_STREAM_DOES_NOT_EXIST);
   }
 
   @Test
+  @BrokerVersionAtLeast("3.11.0")
   void streamInfoFirstOffsetShouldChangeAfterRetentionKickedIn(TestInfo info) {
     int messageCount = 1000;
     int payloadSize = 1000;