Rename "StreamInfo" to "StreamStats"

And use "committed chunk ID" instead of "committed offset".

References rabbitmq/rabbitmq-server#5412

Author: acogoluegnes

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

@@ -68,7 +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_INFO = 28;
+  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

@@ -70,7 +70,7 @@ static EnvironmentBuilder builder() {
    * @return information on the stream
    * @throws UnsupportedOperationException if the broker does not support this command
    */
-  StreamInfo queryStreamInfo(String stream);
+  StreamStats queryStreamInfo(String stream);
 
   /**
    * Create a {@link ProducerBuilder} to configure and create a {@link Producer}.

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

@@ -54,21 +54,21 @@ interface Context {
     long timestamp();
 
     /**
-     * The committed offset in 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.
      *
      * <p>This requires RabbitMQ 3.11 or more. The method always returns 0 otherwise.
      *
-     * @return committed offset in this stream
-     * @see StreamInfo#committedOffset()
+     * @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/StreamInfo.java renamed to src/main/java/com/rabbitmq/stream/StreamStats.java

@@ -20,7 +20,7 @@
  *
  * @see Environment#queryStreamInfo(String)
  */
-public interface StreamInfo {
+public interface StreamStats {
 
   /**
    * The first offset in the stream.
@@ -31,18 +31,18 @@ public interface StreamInfo {
   long firstOffset();
 
   /**
-   * The committed offset in the 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 in this stream
-   * @see Context#committedOffset()
-   * @throws NoOffsetException if there is no committed offset yet
+   * @see Context#committedChunkId()
+   * @throws NoOffsetException if there is no committed chunk yet
    */
-  long committedOffset();
+  long committedChunkId();
 }

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

@@ -32,7 +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_INFO;
+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;
@@ -1355,7 +1355,7 @@ List<FrameHandlerInfo> exchangeCommandVersions() {
     }
   }
 
-  StreamInfoResponse streamInfo(String stream) {
+  StreamInfoResponse streamStats(String stream) {
     if (stream == null) {
       throw new IllegalArgumentException("stream must not be null");
     }
@@ -1364,7 +1364,7 @@ StreamInfoResponse streamInfo(String stream) {
     try {
       ByteBuf bb = allocate(length + 4);
       bb.writeInt(length);
-      bb.writeShort(encodeRequestCode(COMMAND_STREAM_INFO));
+      bb.writeShort(encodeRequestCode(COMMAND_STREAM_STATS));
       bb.writeShort(VERSION_1);
       bb.writeInt(correlationId);
       bb.writeShort(stream.length());
@@ -1473,7 +1473,7 @@ void handle(
         byte subscriptionId,
         long offset,
         long chunkTimestamp,
-        long committedOffset,
+        long committedChunkId,
         Message message);
   }
 
@@ -1890,14 +1890,14 @@ public long getSequence() {
 
   static class StreamInfoResponse extends Response {
 
-    private final Map<String, String> info;
+    private final Map<String, Long> info;
 
-    StreamInfoResponse(short responseCode, Map<String, String> info) {
+    StreamInfoResponse(short responseCode, Map<String, Long> info) {
       super(responseCode);
       this.info = Collections.unmodifiableMap(new HashMap<>(info));
     }
 
-    public Map<String, String> getInfo() {
+    public Map<String, Long> getInfo() {
       return info;
     }
   }

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,7 +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_INFO;
+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;
@@ -134,7 +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_INFO, new StreamInfoFrameHandler());
+    handlers.put(COMMAND_STREAM_STATS, new StreamStatsFrameHandler());
     HANDLERS = new FrameHandler[maxCommandKey + 1][];
     handlers
         .entrySet()
@@ -328,7 +328,7 @@ static int handleMessage(
         long offset,
         long offsetLimit,
         long chunkTimestamp,
-        long committedOffset,
+        long committedChunkId,
         Codec codec,
         MessageListener messageListener,
         byte subscriptionId) {
@@ -342,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;
     }
@@ -600,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
           );
     }
@@ -1150,7 +1150,7 @@ int doHandle(Client client, ChannelHandlerContext ctx, ByteBuf message) {
     }
   }
 
-  private static class StreamInfoFrameHandler extends BaseFrameHandler {
+  private static class StreamStatsFrameHandler extends BaseFrameHandler {
 
     @Override
     int doHandle(Client client, ChannelHandlerContext ctx, ByteBuf message) {
@@ -1162,14 +1162,14 @@ int doHandle(Client client, ChannelHandlerContext ctx, ByteBuf message) {
 
       int infoCount = message.readInt();
       read += 4;
-      Map<String, String> info = new LinkedHashMap<>(infoCount);
+      Map<String, Long> info = new LinkedHashMap<>(infoCount);
 
       for (int i = 0; i < infoCount; i++) {
         String key = readString(message);
         read += 2 + key.length();
-        String value = readString(message);
-        read += 2 + value.length();
+        long value = message.readLong();
         info.put(key, value);
+        read += 8;
       }
 
       OutstandingRequest<StreamInfoResponse> outstandingRequest =

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

@@ -30,7 +30,7 @@
 import com.rabbitmq.stream.ProducerBuilder;
 import com.rabbitmq.stream.StreamCreator;
 import com.rabbitmq.stream.StreamException;
-import com.rabbitmq.stream.StreamInfo;
+import com.rabbitmq.stream.StreamStats;
 import com.rabbitmq.stream.SubscriptionListener;
 import com.rabbitmq.stream.compression.CompressionCodecFactory;
 import com.rabbitmq.stream.impl.Client.ClientParameters;
@@ -390,29 +390,29 @@ public void deleteStream(String stream) {
   }
 
   @Override
-  public StreamInfo queryStreamInfo(String stream) {
+  public StreamStats queryStreamInfo(String stream) {
     StreamInfoResponse response =
         locatorOperation(
             client -> {
               if (Utils.is3_11_OrMore(client.brokerVersion())) {
-                return client.streamInfo(stream);
+                return client.streamStats(stream);
               } else {
                 throw new UnsupportedOperationException(
                     "QueryStringInfo is available only for RabbitMQ 3.11 or more.");
               }
             });
     if (response.isOk()) {
-      Map<String, String> info = response.getInfo();
+      Map<String, Long> info = response.getInfo();
       BiFunction<String, String, LongSupplier> offsetSupplierLogic =
           (key, message) -> {
-            if (!info.containsKey(key) || "-1".equals(info.get(key))) {
+            if (!info.containsKey(key) || info.get(key) == -1) {
               return () -> {
                 throw new NoOffsetException(message);
               };
             } else {
               try {
-                long firstOffset = Long.parseUnsignedLong(info.get(key));
-                return () -> firstOffset;
+                long offset = info.get(key);
+                return () -> offset;
               } catch (NumberFormatException e) {
                 return () -> {
                   throw new NoOffsetException(message);
@@ -421,20 +421,21 @@ public StreamInfo queryStreamInfo(String stream) {
             }
           };
       LongSupplier firstOffsetSupplier =
-          offsetSupplierLogic.apply("first_offset", "No first offset for stream " + stream);
+          offsetSupplierLogic.apply("first_chunk_id", "No first offset for stream " + stream);
       LongSupplier committedOffsetSupplier =
-          offsetSupplierLogic.apply("committed_offset", "No committed offset for stream " + stream);
-      return new DefaultStreamInfo(firstOffsetSupplier, committedOffsetSupplier);
+          offsetSupplierLogic.apply(
+              "committed_chunk_id", "No committed chunk ID for stream " + stream);
+      return new DefaultStreamStats(firstOffsetSupplier, committedOffsetSupplier);
     } else {
       throw propagateException(response.getResponseCode(), stream);
     }
   }
 
-  private static class DefaultStreamInfo implements StreamInfo {
+  private static class DefaultStreamStats implements StreamStats {
 
     private final LongSupplier firstOffsetSupplier, committedOffsetSupplier;
 
-    private DefaultStreamInfo(
+    private DefaultStreamStats(
         LongSupplier firstOffsetSupplier, LongSupplier committedOffsetSupplier) {
       this.firstOffsetSupplier = firstOffsetSupplier;
       this.committedOffsetSupplier = committedOffsetSupplier;
@@ -446,7 +447,7 @@ public long firstOffset() {
     }
 
     @Override
-    public long committedOffset() {
+    public long committedChunkId() {
       return committedOffsetSupplier.getAsLong();
     }
   }

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

@@ -116,8 +116,8 @@ public long timestamp() {
             }
 
             @Override
-            public long committedOffset() {
-              return context.committedOffset();
+            public long committedChunkId() {
+              return context.committedChunkId();
             }
 
             @Override

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

@@ -65,7 +65,6 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 import java.util.function.LongConsumer;
-import java.util.function.ToLongBiFunction;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import org.junit.jupiter.api.Test;
@@ -902,13 +901,13 @@ void streamInfoShouldReturnFirstOffsetAndCommittedOffset() throws Exception {
             new Client.ClientParameters()
                 .chunkListener(chunkListener)
                 .messageListener(messageListener));
-    StreamInfoResponse response = client.streamInfo(stream);
-    assertThat(response.getInfo()).containsEntry("first_offset", "0");
-    assertThat(response.getInfo()).containsEntry("committed_offset", "-1");
+    StreamInfoResponse response = client.streamStats(stream);
+    assertThat(response.getInfo()).containsEntry("first_chunk_id", -1L);
+    assertThat(response.getInfo()).containsEntry("committed_chunk_id", -1L);
     TestUtils.publishAndWaitForConfirms(cf, publishCount, stream);
-    response = client.streamInfo(stream);
-    assertThat(response.getInfo()).containsEntry("first_offset", "0");
-    assertThat(response.getInfo().get("committed_offset")).isNotEqualTo("-1");
+    response = client.streamStats(stream);
+    assertThat(response.getInfo()).containsEntry("first_chunk_id", 0L);
+    assertThat(response.getInfo().get("committed_chunk_id")).isNotEqualTo(-1L);
 
     client.exchangeCommandVersions();
 
@@ -918,13 +917,13 @@ void streamInfoShouldReturnFirstOffsetAndCommittedOffset() throws Exception {
 
     assertThat(latch.await(10, SECONDS)).isTrue();
     assertThat(committedOffset.get()).isPositive();
-    assertThat(committedOffset.toString()).isEqualTo(response.getInfo().get("committed_offset"));
+    assertThat(committedOffset).hasValue(response.getInfo().get("committed_chunk_id"));
   }
 
   @Test
   @BrokerVersionAtLeast("3.11.0")
   void streamInfoShouldReturnErrorWhenStreamDoesNotExist() {
-    assertThat(cf.get().streamInfo("does not exist").getResponseCode())
+    assertThat(cf.get().streamStats("does not exist").getResponseCode())
         .isEqualTo(Constants.RESPONSE_CODE_STREAM_DOES_NOT_EXIST);
   }
 
@@ -947,20 +946,19 @@ void streamInfoFirstOffsetShouldChangeAfterRetentionKickedIn(TestInfo info) {
                   .isOk())
           .isTrue();
 
-      StreamInfoResponse response = client.streamInfo(s);
-      assertThat(response.getInfo()).containsEntry("first_offset", "0");
-      assertThat(response.getInfo()).containsEntry("committed_offset", "-1");
+      StreamInfoResponse response = client.streamStats(s);
+      assertThat(response.getInfo()).containsEntry("first_chunk_id", -1L);
+      assertThat(response.getInfo()).containsEntry("committed_chunk_id", -1L);
 
       byte[] payload = new byte[payloadSize];
       Function<MessageBuilder, Message> messageCreation = mb -> mb.addData(payload).build();
 
       TestUtils.publishAndWaitForConfirms(cf, messageCreation, messageCount, s);
       // publishing again, to make sure new segments trigger retention strategy
       TestUtils.publishAndWaitForConfirms(cf, messageCreation, messageCount, s);
-      response = client.streamInfo(s);
-      ToLongBiFunction<Map<String, String>, String> toOffset = (m, k) -> Long.parseLong(m.get(k));
-      assertThat(toOffset.applyAsLong(response.getInfo(), "first_offset")).isPositive();
-      assertThat(toOffset.applyAsLong(response.getInfo(), "committed_offset")).isPositive();
+      response = client.streamStats(s);
+      assertThat(response.getInfo().get("first_chunk_id")).isPositive();
+      assertThat(response.getInfo().get("committed_chunk_id")).isPositive();
 
     } finally {
       assertThat(client.delete(s).isOk()).isTrue();