Add tests covering new release() call.

apache · Feb 13, 2016 · c9726c2 · c9726c2
1 parent 6134989
commit c9726c2
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 5 deletions.
diff --git a/network/common/src/test/java/org/apache/spark/network/protocol/MessageWithHeaderSuite.java b/network/common/src/test/java/org/apache/spark/network/protocol/MessageWithHeaderSuite.java
@@ -26,9 +26,11 @@
 import io.netty.channel.FileRegion;
 import io.netty.util.AbstractReferenceCounted;
 import org.junit.Test;
+import org.mockito.Mockito;
 
 import static org.junit.Assert.*;
 
+import org.apache.spark.network.TestManagedBuffer;
 import org.apache.spark.network.buffer.ManagedBuffer;
 import org.apache.spark.network.buffer.NettyManagedBuffer;
 import org.apache.spark.network.util.ByteArrayWritableChannel;
@@ -48,14 +50,25 @@ public void testShortWrite() throws Exception {
   @Test
   public void testByteBufBody() throws Exception {
     ByteBuf header = Unpooled.copyLong(42);
-    ByteBuf body = Unpooled.copyLong(84);
+    ByteBuf body = Unpooled.copyLong(84).retain();
     ManagedBuffer managedBuf = new NettyManagedBuffer(body);
     MessageWithHeader msg = new MessageWithHeader(managedBuf, header, body, body.readableBytes());
 
     ByteBuf result = doWrite(msg, 1);
     assertEquals(msg.count(), result.readableBytes());
     assertEquals(42, result.readLong());
     assertEquals(84, result.readLong());
+    msg.deallocate();
+  }
+
+  @Test
+  public void testDeallocateReleasesManagedBuffer() throws Exception {
+    ByteBuf header = Unpooled.copyLong(42);
+    ManagedBuffer managedBuf = Mockito.spy(new TestManagedBuffer(84));
+    ByteBuf body = ((ByteBuf) managedBuf.convertToNetty()).retain();
+    MessageWithHeader msg = new MessageWithHeader(managedBuf, header, body, body.readableBytes());
+    msg.deallocate();
+    Mockito.verify(managedBuf, Mockito.times(1)).release();
   }
 
   private void testFileRegionBody(int totalWrites, int writesPerCall) throws Exception {
@@ -70,6 +83,7 @@ private void testFileRegionBody(int totalWrites, int writesPerCall) throws Excep
     for (long i = 0; i < 8; i++) {
       assertEquals(i, result.readLong());
     }
+    msg.deallocate();
   }
 
   private ByteBuf doWrite(MessageWithHeader msg, int minExpectedWrites) throws Exception {

diff --git a/...ork/common/src/test/java/org/apache/spark/network/server/OneForOneStreamManagerSuite.java b/...ork/common/src/test/java/org/apache/spark/network/server/OneForOneStreamManagerSuite.java
@@ -23,7 +23,6 @@
 import io.netty.channel.Channel;
 import org.junit.Test;
 import org.mockito.Mockito;
-import static org.mockito.Mockito.*;
 
 import org.apache.spark.network.TestManagedBuffer;
 import org.apache.spark.network.buffer.ManagedBuffer;
@@ -40,12 +39,12 @@ public void managedBuffersAreFeedWhenConnectionIsClosed() throws Exception {
     buffers.add(buffer2);
     long streamId = manager.registerStream("appId", buffers.iterator());
 
-    Channel dummyChannel = Mockito.mock(Channel.class);
+    Channel dummyChannel = Mockito.mock(Channel.class, Mockito.RETURNS_SMART_NULLS);
     manager.registerChannel(dummyChannel, streamId);
 
     manager.connectionTerminated(dummyChannel);
 
-    Mockito.verify(buffer1, times(1)).release();
-    Mockito.verify(buffer2, times(1)).release();
+    Mockito.verify(buffer1, Mockito.times(1)).release();
+    Mockito.verify(buffer2, Mockito.times(1)).release();
   }
 }