Add PessimisticallyCalculateMaxWritableSpace

Signed-off-by: Andre Kurait <[email protected]>

TrafficCapture/captureOffloader/src/main/java/org/opensearch/migrations/trafficcapture/StreamChannelConnectionCaptureSerializer.java

@@ -206,6 +206,14 @@ public static int calculateMaxWritableSpace(int totalAvailableSpace, int request
         return Math.min(maxWriteBytesSpace, requestedWriteableSpace);
     }
 
+    // Similar to calculateMaxWritableSpace but perform pessimistic calculation with fewer operations. In some cases returns
+    // up to 1 byte fewer than what could be written out of the available space.
+    public static int pessimisticallyCalculateMaxWritableSpace(int totalAvailableSpace, int requestedWriteableSpace) {
+        final int pessimisticLengthFieldSpace = CodedOutputStream.computeUInt32SizeNoTag(totalAvailableSpace);
+        int maxWriteBytesSpace = totalAvailableSpace - pessimisticLengthFieldSpace;
+        return Math.min(maxWriteBytesSpace, requestedWriteableSpace);
+    }
+
     private void readByteBufIntoCurrentStream(int fieldNum, ByteBuf buf) throws IOException {
         var codedOutputStream = getOrCreateCodedOutputStream();
         if (buf.readableBytes() > 0) {
@@ -346,7 +354,7 @@ private void addDataMessage(int captureFieldNumber, int dataFieldNumber, Instant
             while(readBuffer.readableBytes() > 0) {
                 // addSubstreamMessage will write until COS limit and flush prior if needed
                 spaceLeft = currentOutputStreamWriteableSpaceLeft();
-                var bytesToRead = calculateMaxWritableSpace(spaceLeft - trafficStreamOverhead, readBuffer.readableBytes());
+                var bytesToRead = pessimisticallyCalculateMaxWritableSpace(spaceLeft - trafficStreamOverhead, readBuffer.readableBytes());
                 if (bytesToRead <= 0) {
                     throw new IllegalStateException("Stream space is not allowing forward progress on byteBuf reading");
                 }

TrafficCapture/captureOffloader/src/test/java/org/opensearch/migrations/trafficcapture/StreamChannelConnectionCaptureSerializerTest.java

@@ -538,6 +538,15 @@ public void testCalculateMaxWritableSpace(int totalAvailableSpace, int requested
                                                                 - (calculatedMaxWritableSpace + 1);
             Assertions.assertTrue(expectedSpaceLeftAfterWriteIfOneMoreByteWrote < 0, "expected no space to write one more byte");
         }
+
+        // Test that when PessimisticallyCalculateMaxWritableSpace != calculatedMaxWritableSpace, then
+        // it is positive and equal to calculateMaxWritableSpace - 1
+        var pessimisticWriteableSpace = StreamChannelConnectionCaptureSerializer.pessimisticallyCalculateMaxWritableSpace(totalAvailableSpace,
+            requestedWriteableSpace);
+        if (pessimisticWriteableSpace != calculatedMaxWritableSpace) {
+            Assertions.assertTrue(pessimisticWriteableSpace > 0);
+            Assertions.assertEquals(calculatedMaxWritableSpace - 1, pessimisticWriteableSpace);
+        }
     }
 
     @Test