ARROW-7793: [Java] Release accounted-for reservation memory to parent…

… in case of leak

Closes #6401 from projjal/allcator_leak and squashes the following commits:

c122614 <Projjal Chanda> setting the test to ignore
4abb4e4 <Projjal Chanda> Turn off debug
1573fd5 <Projjal Chanda> Added unit test
0a9f743 <Projjal Chanda> Release accounted-for reservation memory to parent in case of leak

Authored-by: Projjal Chanda <iam@pchanda.com>
Signed-off-by: Pindikura Ravindra <ravindra@dremio.com>

java/memory/src/main/java/org/apache/arrow/memory/BaseAllocator.java

@@ -479,8 +479,11 @@ public synchronized void close() {
     // Is there unaccounted-for outstanding allocation?
     final long allocated = getAllocatedMemory();
     if (allocated > 0) {
-      String msg = String.format("Memory was leaked by query. Memory leaked: (%d)\n%s", allocated,
-          toString());
+      if (parent != null && reservation > allocated) {
+        parent.releaseBytes(reservation - allocated);
+      }
+      String msg = String.format("Memory was leaked by query. Memory leaked: (%d)\n%s%s", allocated,
+          outstandingChildAllocators.toString(), toString());
       logger.error(msg);
       throw new IllegalStateException(msg);
     }

java/memory/src/test/java/org/apache/arrow/memory/TestBaseAllocator.java

@@ -35,6 +35,7 @@
 import org.apache.arrow.memory.AllocationOutcomeDetails.Entry;
 import org.apache.arrow.memory.rounding.RoundingPolicy;
 import org.apache.arrow.memory.rounding.SegmentRoundingPolicy;
+import org.apache.arrow.memory.util.AssertionUtil;
 import org.junit.Ignore;
 import org.junit.Test;
 import org.junit.jupiter.api.Assertions;
@@ -1132,6 +1133,40 @@ public void multiple() throws Exception {
     }
   }
 
+  // This test needs to run in non-debug mode. So disabling the assertion status through class loader for this.
+  // The test passes if run individually with -Dtest=TestBaseAllocator#testMemoryLeakWithReservation
+  // but fails generally since the assertion status cannot be changed once the class is initialized.
+  // So setting the test to @ignore
+  @Test(expected = IllegalStateException.class)
+  @Ignore
+  public void testMemoryLeakWithReservation() throws Exception {
+    // disabling assertion status
+    AssertionUtil.class.getClassLoader().setClassAssertionStatus(AssertionUtil.class.getName(), false);
+    try (RootAllocator rootAllocator = new RootAllocator(MAX_ALLOCATION)) {
+      ChildAllocator childAllocator1 = (ChildAllocator) rootAllocator.newChildAllocator(
+          "child1", 1024, MAX_ALLOCATION);
+      rootAllocator.verify();
+
+      ChildAllocator childAllocator2 = (ChildAllocator) childAllocator1.newChildAllocator(
+          "child2", 1024, MAX_ALLOCATION);
+      rootAllocator.verify();
+
+      ArrowBuf buff = childAllocator2.buffer(256);
+
+      Exception exception = assertThrows(IllegalStateException.class, () -> {
+        childAllocator2.close();
+      });
+      String exMessage = exception.getMessage();
+      assertTrue(exMessage.contains("Memory leaked: (256)"));
+
+      exception = assertThrows(IllegalStateException.class, () -> {
+        childAllocator1.close();
+      });
+      exMessage = exception.getMessage();
+      assertTrue(exMessage.contains("Memory leaked: (256)"));
+    }
+  }
+
   public void assertEquiv(ArrowBuf origBuf, ArrowBuf newBuf) {
     assertEquals(origBuf.readerIndex(), newBuf.readerIndex());
     assertEquals(origBuf.writerIndex(), newBuf.writerIndex());