BBuf indexes are ints, not longs.

Motivation:
 We don't intend on support buffers bigger than 2 GiB, which is the max IO transfer size
 on Linux.

Modification:
 Change buffer read/write indexes to be ints, and add checks to allocators.

Result:
 BBuf is now int-sized and indexed.

buffer/src/main/java/io/netty/buffer/b2/Allocator.java

@@ -4,7 +4,20 @@ import jdk.incubator.foreign.MemorySegment;
 
 import static io.netty.buffer.b2.BBuf.*;
 
+@SuppressWarnings("InterfaceMayBeAnnotatedFunctional")
 public interface Allocator extends AutoCloseable {
+    static void checkSize(long size) {
+        if (size < 1) {
+            throw new IllegalArgumentException("Buffer size must be positive, but was " + size + '.');
+        }
+        // We use max array size because on-heap buffers will be backed by byte-arrays.
+        int maxArraySize = Integer.MAX_VALUE - 8;
+        if (size > maxArraySize) {
+            throw new IllegalArgumentException(
+                    "Buffer size cannot be greater than " + maxArraySize + ", but was " + size + '.');
+        }
+    }
+
     BBuf allocate(long size);
 
     @Override
@@ -15,6 +28,7 @@ public interface Allocator extends AutoCloseable {
         return new Allocator() {
             @Override
             public BBuf allocate(long size) {
+                checkSize(size);
                 var segment = MemorySegment.ofArray(new byte[Math.toIntExact(size)]);
                 return new BBuf(segment, SEGMENT_CLOSE);
             }
@@ -25,6 +39,7 @@ public interface Allocator extends AutoCloseable {
         return new Allocator() {
             @Override
             public BBuf allocate(long size) {
+                checkSize(size);
                 var segment = MemorySegment.allocateNative(size);
                 Statics.MEM_USAGE_NATIVE.add(size);
                 return new BBuf(segment, SEGMENT_CLOSE_NATIVE);
@@ -36,6 +51,7 @@ public interface Allocator extends AutoCloseable {
         return new SizeClassedMemoryPool(false) {
             @Override
             protected MemorySegment createMemorySegment(long size) {
+                checkSize(size);
                 return MemorySegment.ofArray(new byte[Math.toIntExact(size)]);
             }
         };
@@ -45,6 +61,7 @@ public interface Allocator extends AutoCloseable {
         return new SizeClassedMemoryPool(true) {
             @Override
             protected MemorySegment createMemorySegment(long size) {
+                checkSize(size);
                 return MemorySegment.allocateNative(size);
             }
         };
@@ -54,6 +71,7 @@ public interface Allocator extends AutoCloseable {
         return new SizeClassedMemoryPool(true) {
             @Override
             protected MemorySegment createMemorySegment(long size) {
+                checkSize(size);
                 return MemorySegment.allocateNative(size);
             }
 

buffer/src/main/java/io/netty/buffer/b2/BBuf.java

@@ -15,15 +15,15 @@ public class BBuf extends Rc<BBuf> {
         MEM_USAGE_NATIVE.add(-buf.segment.byteSize());
     };
     final MemorySegment segment;
-    private long read;
-    private long write;
+    private int read;
+    private int write;
 
     BBuf(MemorySegment segment, Drop<BBuf> drop) {
         super(drop);
         this.segment = segment;
     }
 
-    public BBuf readerIndex(long index) {
+    public BBuf readerIndex(int index) {
         read = index;
         return this;
     }
@@ -40,12 +40,12 @@ public class BBuf extends Rc<BBuf> {
         MemoryAccess.setByteAtOffset(segment, write++, value);
     }
 
-    public BBuf setLong(long offset, long value) {
+    public BBuf setLong(int offset, long value) {
         MemoryAccess.setLongAtOffset(segment, offset, value);
         return this;
     }
 
-    public long getLong(long offset) {
+    public long getLong(int offset) {
         return MemoryAccess.getLongAtOffset(segment, offset);
     }
 

buffer/src/test/java/io/netty/buffer/b2/BBufTest.java

@@ -1,5 +1,6 @@
 package io.netty.buffer.b2;
 
+import org.junit.AssumptionViolatedException;
 import org.junit.Test;
 
 import java.util.concurrent.ArrayBlockingQueue;
@@ -147,5 +148,50 @@ public abstract class BBufTest {
             assertEquals((byte) 42, future.get().byteValue());
         }
     }
+
+    @Test
+    public void mustThrowWhenAllocatingZeroSizedBuffer() {
+        try (Allocator allocator = createAllocator()) {
+            try {
+                allocator.allocate(0);
+                fail("Expected to throw an IllegalArgumentException.");
+            } catch (IllegalArgumentException ignore) {
+            }
+        }
+    }
+
+    @Test
+    public void mustThrowWhenAllocatingNegativeSizedBuffer() {
+        try (Allocator allocator = createAllocator()) {
+            try {
+                allocator.allocate(-1);
+                fail("Expected to throw an IllegalArgumentException.");
+            } catch (IllegalArgumentException ignore) {
+            }
+        }
+    }
+
+    @Test
+    public void mustThrowWhenAllocatingOverSizedBuffer() {
+        try (Allocator allocator = createAllocator()) {
+            try {
+                allocator.allocate(Integer.MAX_VALUE);
+                fail("Expected to throw an IllegalArgumentException.");
+            } catch (IllegalArgumentException ignore) {
+            }
+        }
+    }
+
+    @Test
+    public void mustAllowAllocatingMaxArraySizedBuffer() {
+        try (Allocator allocator = createAllocator()) {
+            try {
+                allocator.allocate(Integer.MAX_VALUE - 8).close();
+            } catch (OutOfMemoryError oome) {
+                // Mark test as ignored if this happens.
+                throw new AssumptionViolatedException("JVM does not have enough memory for this test.", oome);
+            }
+        }
+    }
 }