Support offset type field in resolve_offset_spec response

References #928

Author: acogoluegnes

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

@@ -1,4 +1,4 @@
-// Copyright (c) 2020-2025 Broadcom. All Rights Reserved.
+// Copyright (c) 2020-2026 Broadcom. All Rights Reserved.
 // The term "Broadcom" refers to Broadcom Inc. and/or its subsidiaries.
 //
 // This software, the RabbitMQ Stream Java client library, is dual-licensed under the

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

@@ -2425,13 +2425,19 @@ public long getOffset() {
 
   public static class ResolveOffsetSpecResponse extends Response {
 
+    private final short offsetType;
     private final long offset;
 
-    ResolveOffsetSpecResponse(short responseCode, long offset) {
+    ResolveOffsetSpecResponse(short responseCode, short offsetType, long offset) {
       super(responseCode);
+      this.offsetType = offsetType;
       this.offset = offset;
     }
 
+    public short offsetType() {
+      return offsetType;
+    }
+
     public long offset() {
       return offset;
     }

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

@@ -1273,6 +1273,8 @@ int doHandle(Client client, ChannelHandlerContext ctx, ByteBuf message) {
       int read = 4;
       short responseCode = message.readShort();
       read += 2;
+      short offsetType = message.readShort();
+      read += 2;
       long offset = message.readLong();
       read += 8;
 
@@ -1281,7 +1283,8 @@ int doHandle(Client client, ChannelHandlerContext ctx, ByteBuf message) {
       if (outstandingRequest == null) {
         logMissingOutstandingRequest(correlationId);
       } else {
-        ResolveOffsetSpecResponse response = new ResolveOffsetSpecResponse(responseCode, offset);
+        ResolveOffsetSpecResponse response =
+            new ResolveOffsetSpecResponse(responseCode, offsetType, offset);
         outstandingRequest.response().set(response);
         outstandingRequest.countDown();
       }

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

@@ -1146,18 +1146,22 @@ void resolveOffsetSpecShouldReturnResolvedOffset() {
 
   static void doResolveOffsetSpecShouldReturnResolvedOffset(
       TestUtils.ClientFactory cf, Client client, String stream) {
+    short typeOffset = OffsetSpecification.offset(0).getType();
     // Test resolve_offset_spec on empty stream
     Client.ResolveOffsetSpecResponse response =
         client.resolveOffsetSpec(stream, OffsetSpecification.first());
     assertThat(response).is(ok());
+    assertThat(response.offsetType()).isEqualTo(typeOffset);
     assertThat(response.offset()).isEqualTo(0);
 
     response = client.resolveOffsetSpec(stream, OffsetSpecification.last());
     assertThat(response).is(ok());
+    assertThat(response.offsetType()).isEqualTo(typeOffset);
     assertThat(response.offset()).isEqualTo(0);
 
     response = client.resolveOffsetSpec(stream, OffsetSpecification.next());
     assertThat(response).is(ok());
+    assertThat(response.offsetType()).isEqualTo(typeOffset);
     assertThat(response.offset()).isEqualTo(0);
 
     // Publish some messages
@@ -1167,34 +1171,41 @@ static void doResolveOffsetSpecShouldReturnResolvedOffset(
     // Test resolve_offset_spec after publishing
     response = client.resolveOffsetSpec(stream, OffsetSpecification.first());
     assertThat(response).is(ok());
+    assertThat(response.offsetType()).isEqualTo(typeOffset);
     assertThat(response.offset()).isEqualTo(0);
 
     response = client.resolveOffsetSpec(stream, OffsetSpecification.last());
     assertThat(response).is(ok());
+    assertThat(response.offsetType()).isEqualTo(typeOffset);
     assertThat(response.offset()).isGreaterThanOrEqualTo(0);
     assertThat(response.offset()).isLessThan(messageCount);
 
     response = client.resolveOffsetSpec(stream, OffsetSpecification.next());
     assertThat(response).is(ok());
+    assertThat(response.offsetType()).isEqualTo(typeOffset);
     assertThat(response.offset()).isEqualTo(messageCount);
 
     response = client.resolveOffsetSpec(stream, OffsetSpecification.offset(0));
     assertThat(response).is(ok());
+    assertThat(response.offsetType()).isEqualTo(typeOffset);
     assertThat(response.offset()).isEqualTo(0);
 
     response = client.resolveOffsetSpec(stream, OffsetSpecification.offset(5));
     assertThat(response).is(ok());
+    assertThat(response.offsetType()).isEqualTo(typeOffset);
     // it should attach at the beginning of the chunk
     assertThat(response.offset()).isLessThanOrEqualTo(5);
 
     // Test with timestamp (far future should return next offset)
     long futureTimestamp = System.currentTimeMillis() + 3600000;
     response = client.resolveOffsetSpec(stream, OffsetSpecification.timestamp(futureTimestamp));
     assertThat(response).is(ok());
+    assertThat(response.offsetType()).isEqualTo(typeOffset);
     assertThat(response.offset()).isEqualTo(messageCount);
 
     response = client.resolveOffsetSpec(stream, OffsetSpecification.offset(messageCount * 2));
     assertThat(response).is(ok());
+    assertThat(response.offsetType()).isEqualTo(typeOffset);
     assertThat(response.offset()).isEqualTo(messageCount);
 
     // Test on non-existent stream