Merge pull request #165 from rabbitmq/rabbitmq-server-5412-stream-info-command

Support StreamInfo command

Author: acogoluegnes

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

@@ -68,6 +68,7 @@ public final class Constants {
   public static final short COMMAND_ROUTE = 24;
   public static final short COMMAND_PARTITIONS = 25;
   public static final short COMMAND_EXCHANGE_COMMAND_VERSIONS = 27;
+  public static final short COMMAND_STREAM_STATS = 28;
 
   public static final short VERSION_1 = 1;
   public static final short VERSION_2 = 2;

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2022 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").
@@ -61,6 +61,17 @@ 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
+   */
+  StreamStats queryStreamInfo(String stream);
+
   /**
    * Create a {@link ProducerBuilder} to configure and create a {@link Producer}.
    *

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2022 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").
@@ -54,18 +54,21 @@ interface Context {
     long timestamp();
 
     /**
-     * The committed offset on this stream.
+     * The ID (offset) of the committed chunk (block of messages) 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>It is the offset of the first message in the last chunk 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.
+     * <p>The committed chunk ID is a good indication of what the last offset of a stream can be at
+     * a given time. The value can be stale as soon as the application reads it though, as the
+     * committed chunk ID 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 chunk ID in this stream
+     * @see StreamStats#committedChunkId()
      */
-    long committedOffset();
+    long committedChunkId();
 
     /**
      * The consumer that receives the message.

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

@@ -0,0 +1,21 @@
+// Copyright (c) 2022 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;
+
+public class NoOffsetException extends StreamException {
+
+  public NoOffsetException(String message) {
+    super(message, Constants.RESPONSE_CODE_NO_OFFSET);
+  }
+}

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2021 VMware, Inc. or its affiliates.  All rights reserved.
+// Copyright (c) 2020-2022 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").
@@ -19,7 +19,7 @@ public class StreamDoesNotExistException extends StreamException {
   private final String stream;
 
   public StreamDoesNotExistException(String stream) {
-    super("Stream " + stream + "  does not exist");
+    super("Stream " + stream + "  does not exist", Constants.RESPONSE_CODE_STREAM_DOES_NOT_EXIST);
     this.stream = stream;
   }
 

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

@@ -0,0 +1,22 @@
+// Copyright (c) 2022 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;
+
+/** Exception to indicate a stream is not available. */
+public class StreamNotAvailableException extends StreamException {
+
+  public StreamNotAvailableException(String stream) {
+    super("Stream " + stream + " is not available", Constants.RESPONSE_CODE_STREAM_NOT_AVAILABLE);
+  }
+}

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

@@ -0,0 +1,48 @@
+// Copyright (c) 2022 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;
+
+import com.rabbitmq.stream.MessageHandler.Context;
+
+/**
+ * Information on a stream.
+ *
+ * @see Environment#queryStreamInfo(String)
+ */
+public interface StreamStats {
+
+  /**
+   * The first offset in the stream.
+   *
+   * @return first offset in the stream
+   * @throws NoOffsetException if there is no first offset yet
+   */
+  long firstOffset();
+
+  /**
+   * The ID (offset) of the committed chunk (block of messages) in the stream.
+   *
+   * <p>It is the offset of the first message in the last chunk confirmed by a quorum of the stream
+   * cluster members (leader and replicas).
+   *
+   * <p>The committed chunk ID is a good indication of what the last offset of a stream can be at a
+   * given time. The value can be stale as soon as the application reads it though, as the committed
+   * chunk ID for a stream that is published to changes all the time.
+   *
+   * @return committed offset in this stream
+   * @see Context#committedChunkId()
+   * @throws NoOffsetException if there is no committed chunk yet
+   */
+  long committedChunkId();
+}

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

@@ -32,6 +32,7 @@
 import static com.rabbitmq.stream.Constants.COMMAND_SASL_AUTHENTICATE;
 import static com.rabbitmq.stream.Constants.COMMAND_SASL_HANDSHAKE;
 import static com.rabbitmq.stream.Constants.COMMAND_STORE_OFFSET;
+import static com.rabbitmq.stream.Constants.COMMAND_STREAM_STATS;
 import static com.rabbitmq.stream.Constants.COMMAND_SUBSCRIBE;
 import static com.rabbitmq.stream.Constants.COMMAND_UNSUBSCRIBE;
 import static com.rabbitmq.stream.Constants.RESPONSE_CODE_AUTHENTICATION_FAILURE;
@@ -1354,6 +1355,31 @@ List<FrameHandlerInfo> exchangeCommandVersions() {
     }
   }
 
+  StreamInfoResponse streamStats(String stream) {
+    if (stream == null) {
+      throw new IllegalArgumentException("stream must not be null");
+    }
+    int length = 2 + 2 + 4 + 2 + stream.length(); // API code, version, correlation ID, 1 string
+    int correlationId = correlationSequence.incrementAndGet();
+    try {
+      ByteBuf bb = allocate(length + 4);
+      bb.writeInt(length);
+      bb.writeShort(encodeRequestCode(COMMAND_STREAM_STATS));
+      bb.writeShort(VERSION_1);
+      bb.writeInt(correlationId);
+      bb.writeShort(stream.length());
+      bb.writeBytes(stream.getBytes(StandardCharsets.UTF_8));
+      OutstandingRequest<StreamInfoResponse> request = new OutstandingRequest<>(this.rpcTimeout);
+      outstandingRequests.put(correlationId, request);
+      channel.writeAndFlush(bb);
+      request.block();
+      return request.response.get();
+    } catch (RuntimeException e) {
+      outstandingRequests.remove(correlationId);
+      throw new StreamException(e);
+    }
+  }
+
   void shutdownReason(ShutdownReason reason) {
     this.shutdownReason = reason;
   }
@@ -1447,7 +1473,7 @@ void handle(
         byte subscriptionId,
         long offset,
         long chunkTimestamp,
-        long committedOffset,
+        long committedChunkId,
         Message message);
   }
 
@@ -1852,7 +1878,7 @@ static class QueryPublisherSequenceResponse extends Response {
 
     private final long sequence;
 
-    public QueryPublisherSequenceResponse(short responseCode, long sequence) {
+    QueryPublisherSequenceResponse(short responseCode, long sequence) {
       super(responseCode);
       this.sequence = sequence;
     }
@@ -1862,6 +1888,20 @@ public long getSequence() {
     }
   }
 
+  static class StreamInfoResponse extends Response {
+
+    private final Map<String, Long> info;
+
+    StreamInfoResponse(short responseCode, Map<String, Long> info) {
+      super(responseCode);
+      this.info = Collections.unmodifiableMap(new HashMap<>(info));
+    }
+
+    public Map<String, Long> getInfo() {
+      return info;
+    }
+  }
+
   public static class StreamMetadata {
 
     private final String stream;

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

@@ -310,7 +310,7 @@ public long timestamp() {
     }
 
     @Override
-    public long committedOffset() {
+    public long committedChunkId() {
       return committedOffset;
     }
 

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

@@ -34,6 +34,7 @@
 import static com.rabbitmq.stream.Constants.COMMAND_ROUTE;
 import static com.rabbitmq.stream.Constants.COMMAND_SASL_AUTHENTICATE;
 import static com.rabbitmq.stream.Constants.COMMAND_SASL_HANDSHAKE;
+import static com.rabbitmq.stream.Constants.COMMAND_STREAM_STATS;
 import static com.rabbitmq.stream.Constants.COMMAND_SUBSCRIBE;
 import static com.rabbitmq.stream.Constants.COMMAND_TUNE;
 import static com.rabbitmq.stream.Constants.COMMAND_UNSUBSCRIBE;
@@ -62,6 +63,7 @@
 import com.rabbitmq.stream.impl.Client.Response;
 import com.rabbitmq.stream.impl.Client.SaslAuthenticateResponse;
 import com.rabbitmq.stream.impl.Client.ShutdownContext.ShutdownReason;
+import com.rabbitmq.stream.impl.Client.StreamInfoResponse;
 import com.rabbitmq.stream.impl.Client.StreamMetadata;
 import com.rabbitmq.stream.impl.Client.SubscriptionOffset;
 import com.rabbitmq.stream.metrics.MetricsCollector;
@@ -132,6 +134,7 @@ class ServerFrameHandler {
     handlers.put(COMMAND_ROUTE, new RouteFrameHandler());
     handlers.put(COMMAND_PARTITIONS, new PartitionsFrameHandler());
     handlers.put(COMMAND_EXCHANGE_COMMAND_VERSIONS, new ExchangeCommandVersionsFrameHandler());
+    handlers.put(COMMAND_STREAM_STATS, new StreamStatsFrameHandler());
     HANDLERS = new FrameHandler[maxCommandKey + 1][];
     handlers
         .entrySet()
@@ -325,7 +328,7 @@ static int handleMessage(
         long offset,
         long offsetLimit,
         long chunkTimestamp,
-        long committedOffset,
+        long committedChunkId,
         Codec codec,
         MessageListener messageListener,
         byte subscriptionId) {
@@ -339,7 +342,7 @@ static int handleMessage(
         messageFiltered.set(true);
       } else {
         Message message = codec.decode(data);
-        messageListener.handle(subscriptionId, offset, chunkTimestamp, committedOffset, message);
+        messageListener.handle(subscriptionId, offset, chunkTimestamp, committedChunkId, message);
       }
       return read;
     }
@@ -597,7 +600,7 @@ int doHandle(Client client, ChannelHandlerContext ctx, ByteBuf message) {
           client.chunkChecksum,
           client.metricsCollector,
           message.readByte(), // subscription ID
-          message.readLong(), // last committed offset, unsigned long
+          message.readLong(), // committed chunk ID, unsigned long
           9 // byte read count, 1 + 9
           );
     }
@@ -1146,4 +1149,38 @@ int doHandle(Client client, ChannelHandlerContext ctx, ByteBuf message) {
       return read;
     }
   }
+
+  private static class StreamStatsFrameHandler extends BaseFrameHandler {
+
+    @Override
+    int doHandle(Client client, ChannelHandlerContext ctx, ByteBuf message) {
+      int correlationId = message.readInt();
+      int read = 4;
+
+      short responseCode = message.readShort();
+      read += 2;
+
+      int infoCount = message.readInt();
+      read += 4;
+      Map<String, Long> info = new LinkedHashMap<>(infoCount);
+
+      for (int i = 0; i < infoCount; i++) {
+        String key = readString(message);
+        read += 2 + key.length();
+        long value = message.readLong();
+        info.put(key, value);
+        read += 8;
+      }
+
+      OutstandingRequest<StreamInfoResponse> outstandingRequest =
+          remove(client.outstandingRequests, correlationId, StreamInfoResponse.class);
+      if (outstandingRequest == null) {
+        LOGGER.warn("Could not find outstanding request with correlation ID {}", correlationId);
+      } else {
+        outstandingRequest.response().set(new StreamInfoResponse(responseCode, info));
+        outstandingRequest.countDown();
+      }
+      return read;
+    }
+  }
 }