netty: Eliminate buffer release when there is an error as caller still owns the buffer. (#10537)

* netty: Eliminate buffer release when there is an error as caller still owns the buffer.
---------

Co-authored-by: Sergii Tkachenko <hi@sergii.org>

netty/src/main/java/io/grpc/netty/NettyAdaptiveCumulator.java

@@ -204,16 +204,11 @@ class NettyAdaptiveCumulator implements Cumulator {
       composite.addFlattenedComponents(true, newTail);
       // New tail's ownership transferred to the composite buf.
       newTail = null;
-      in.release();
-      in = null;
-      // Restore the reader. In case it fails we restore the reader after releasing/forgetting
-      // the input and the new tail so that finally block can handles them properly.
       composite.readerIndex(prevReader);
+      // Input buffer was successfully merged with the tail.
+      // Must be the last line in the try because we release it only on success.
+      in.release();
     } finally {
-      // Input buffer was merged with the tail.
-      if (in != null) {
-        in.release();
-      }
       // If new tail's ownership isn't transferred to the composite buf.
       // Release it to prevent a leak.
       if (newTail != null) {

netty/src/test/java/io/grpc/netty/NettyAdaptiveCumulatorTest.java

@@ -533,11 +533,12 @@ public class NettyAdaptiveCumulatorTest {
 
       try {
         NettyAdaptiveCumulator.mergeWithCompositeTail(alloc, compositeThrows, in);
+        in = null; // On success it would be released
         fail("Cumulator didn't throw");
       } catch (UnsupportedOperationException actualError) {
         assertSame(expectedError, actualError);
-        // Input must be released unless its ownership has been to the composite cumulation.
+        // Because of error, ownership shouldn't have changed so should not have been released.
-        assertEquals(0, in.refCnt());
+        assertEquals(1, in.refCnt());
         // Tail released
         assertEquals(0, tail.refCnt());
         // Composite cumulation is retained
@@ -545,6 +546,9 @@ public class NettyAdaptiveCumulatorTest {
         // Composite cumulation loses the tail
         assertEquals(0, compositeThrows.numComponents());
       } finally {
+        if (in != null) {
+          in.release();
+        }
         compositeThrows.release();
       }
     }
@@ -569,11 +573,12 @@ public class NettyAdaptiveCumulatorTest {
 
       try {
         NettyAdaptiveCumulator.mergeWithCompositeTail(mockAlloc, compositeRo, in);
+        in = null; // On success it would be released
         fail("Cumulator didn't throw");
       } catch (UnsupportedOperationException actualError) {
         assertSame(expectedError, actualError);
-        // Input must be released unless its ownership has been to the composite cumulation.
+        // Because of error, ownership shouldn't have changed so should not have been released.
-        assertEquals(0, in.refCnt());
+        assertEquals(1, in.refCnt());
         // New buffer released
         assertEquals(0, newTail.refCnt());
         // Composite cumulation is retained
@@ -581,6 +586,9 @@ public class NettyAdaptiveCumulatorTest {
         // Composite cumulation loses the tail
         assertEquals(0, compositeRo.numComponents());
       } finally {
+        if (in != null) {
+          in.release();
+        }
         compositeRo.release();
       }
     }