ByteBuffer is not released when compression is enabled, causing pool leakage
@@ -724,7 +724,8 @@ public class InternalStreamConnection implements InternalConnection {
}
ByteBuf messageBuffer = stream.read(messageHeader.getMessageLength() - MESSAGE_HEADER_LENGTH, additionalTimeout);
-
+ boolean releaseMessageBuffer = true;
+ try {
if (messageHeader.getOpCode() == OP_COMPRESSED.getValue()) {
CompressedHeader compressedHeader = new CompressedHeader(messageBuffer, messageHeader);
@@ -736,7 +737,14 @@ public class InternalStreamConnection implements InternalConnection {
buffer.flip();
return new ResponseBuffers(new ReplyHeader(buffer, compressedHeader), buffer);
} else {
- return new ResponseBuffers(new ReplyHeader(messageBuffer, messageHeader), messageBuffer);
+ ResponseBuffers responseBuffers = new ResponseBuffers(new ReplyHeader(messageBuffer, messageHeader), messageBuffer);
+ releaseMessageBuffer = false;
+ return responseBuffers;
+ }
+ } finally {
+ if (releaseMessageBuffer) {
+ messageBuffer.release();
+ }
}
}
@@ -792,6 +800,7 @@ public class InternalStreamConnection implements InternalConnection {
callback.onResult(null, t);
return;
}
+ boolean releaseResult = true;
try {
ReplyHeader replyHeader;
ByteBuf responseBuffer;
@@ -806,15 +815,21 @@ public class InternalStreamConnection implements InternalConnection {
replyHeader = new ReplyHeader(buffer, compressedHeader);
responseBuffer = buffer;
} finally {
+ releaseResult = false;
result.release();
}
} else {
replyHeader = new ReplyHeader(result, messageHeader);
responseBuffer = result;
+ releaseResult = false;
}
callback.onResult(new ResponseBuffers(replyHeader, responseBuffer), null);
} catch (Throwable localThrowable) {
callback.onResult(null, localThrowable);
+ } finally {
+ if (releaseResult) {
+ result.release();
+ }
}
}
}
