You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@pekko.apache.org by "He-Pin (via GitHub)" <gi...@apache.org> on 2024/12/28 14:30:47 UTC
[PR] fix: Fix ByteBuf leak in remoting transport. [pekko]
He-Pin opened a new pull request, #1635:
URL: https://github.com/apache/pekko/pull/1635
Motivation:
refs: https://github.com/apache/pekko/issues/1634
Modification:
Call `byteBuf.release` after bytes read.
Result:
No leaks.
Sorry for everyone.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: Fix ByteBuf leak in remoting transport. [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on PR #1635:
URL: https://github.com/apache/pekko/pull/1635#issuecomment-2564373345
OK, Will split
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: Fix ByteBuf leak in remoting transport. [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1635:
URL: https://github.com/apache/pekko/pull/1635#discussion_r1898944106
##########
remote/src/main/scala/org/apache/pekko/remote/transport/netty/TcpSupport.scala:
##########
@@ -56,8 +58,16 @@ private[remote] trait TcpHandlers extends CommonHandlers {
}
override def onMessage(ctx: ChannelHandlerContext, msg: ByteBuf): Unit = {
- val bytes: Array[Byte] = ByteBufUtil.getBytes(msg)
- if (bytes.length > 0) notifyListener(ctx.channel(), InboundPayload(ByteString(bytes)))
+ try {
+ val bytes: Array[Byte] = ByteBufUtil.getBytes(msg)
+ if (bytes.length > 0) notifyListener(ctx.channel(), InboundPayload(ByteString(bytes)))
+ } catch {
+ case NonFatal(e) =>
+ log.warning("Error while handling message from [{}]: {}", ctx.channel().remoteAddress(), e)
+ onException(ctx, e)
+ } finally {
+ msg.release()
Review Comment:
I can't call release, because it's a `retainedSlice` not a `readRetainedSlice`, will dig it deeper
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: Fix ByteBuf leak in remoting transport. [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #1635:
URL: https://github.com/apache/pekko/pull/1635#discussion_r1898915474
##########
multi-node-testkit/src/main/scala/org/apache/pekko/remote/testconductor/RemoteConnection.scala:
##########
@@ -56,8 +56,15 @@ private[pekko] class ProtobufEncoder extends MessageToMessageEncoder[Message] {
private[pekko] class ProtobufDecoder(prototype: Message) extends MessageToMessageDecoder[ByteBuf] {
override def decode(ctx: ChannelHandlerContext, msg: ByteBuf, out: java.util.List[AnyRef]): Unit = {
- val bytes = ByteBufUtil.getBytes(msg)
- out.add(prototype.getParserForType.parseFrom(bytes))
+ try {
+ val bytes = ByteBufUtil.getBytes(msg)
+ out.add(prototype.getParserForType.parseFrom(bytes))
+ } catch {
+ case NonFatal(e) =>
+ throw new RuntimeException("Error while decoding protobuf message", e)
Review Comment:
Can you suggest to Flink team to revert their upgrade to Pekko 1.1 if you think that this is definietly broken.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: Fix ByteBuf leak in remoting transport. [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1635:
URL: https://github.com/apache/pekko/pull/1635#discussion_r1898916610
##########
remote/src/main/scala/org/apache/pekko/remote/transport/netty/TcpSupport.scala:
##########
@@ -56,8 +58,16 @@ private[remote] trait TcpHandlers extends CommonHandlers {
}
override def onMessage(ctx: ChannelHandlerContext, msg: ByteBuf): Unit = {
- val bytes: Array[Byte] = ByteBufUtil.getBytes(msg)
- if (bytes.length > 0) notifyListener(ctx.channel(), InboundPayload(ByteString(bytes)))
+ try {
+ val bytes: Array[Byte] = ByteBufUtil.getBytes(msg)
+ if (bytes.length > 0) notifyListener(ctx.channel(), InboundPayload(ByteString(bytes)))
+ } catch {
+ case NonFatal(e) =>
+ log.warning("Error while handling message from [{}]: {}", ctx.channel().remoteAddress(), e)
+ onException(ctx, e)
+ } finally {
+ msg.release()
Review Comment:
Sorry for not checking the `LengthFieldBasedFrameDecoder` in detail, we need to turn on the leak detector in nightly testing I think.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: Fix ByteBuf leak in remoting transport. [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1635:
URL: https://github.com/apache/pekko/pull/1635#discussion_r1898917450
##########
multi-node-testkit/src/main/scala/org/apache/pekko/remote/testconductor/RemoteConnection.scala:
##########
@@ -56,8 +56,15 @@ private[pekko] class ProtobufEncoder extends MessageToMessageEncoder[Message] {
private[pekko] class ProtobufDecoder(prototype: Message) extends MessageToMessageDecoder[ByteBuf] {
override def decode(ctx: ChannelHandlerContext, msg: ByteBuf, out: java.util.List[AnyRef]): Unit = {
- val bytes = ByteBufUtil.getBytes(msg)
- out.add(prototype.getParserForType.parseFrom(bytes))
+ try {
+ val bytes = ByteBufUtil.getBytes(msg)
+ out.add(prototype.getParserForType.parseFrom(bytes))
+ } catch {
+ case NonFatal(e) =>
+ throw new RuntimeException("Error while decoding protobuf message", e)
Review Comment:
@pjfanning I have notified with an Email.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: Fix ByteBuf leak in remoting transport. [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #1635:
URL: https://github.com/apache/pekko/pull/1635#discussion_r1898915322
##########
multi-node-testkit/src/main/scala/org/apache/pekko/remote/testconductor/RemoteConnection.scala:
##########
@@ -56,8 +56,15 @@ private[pekko] class ProtobufEncoder extends MessageToMessageEncoder[Message] {
private[pekko] class ProtobufDecoder(prototype: Message) extends MessageToMessageDecoder[ByteBuf] {
override def decode(ctx: ChannelHandlerContext, msg: ByteBuf, out: java.util.List[AnyRef]): Unit = {
- val bytes = ByteBufUtil.getBytes(msg)
- out.add(prototype.getParserForType.parseFrom(bytes))
+ try {
+ val bytes = ByteBufUtil.getBytes(msg)
+ out.add(prototype.getParserForType.parseFrom(bytes))
+ } catch {
+ case NonFatal(e) =>
+ throw new RuntimeException("Error while decoding protobuf message", e)
Review Comment:
why not rethrow the e? I dislike using the super generic RuntimeException here. If we need to wrap the exception, can't we find something more specific or create a new class like NettyDecodeException.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: Fix ByteBuf leak in remoting transport. [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on code in PR #1635:
URL: https://github.com/apache/pekko/pull/1635#discussion_r1898915474
##########
multi-node-testkit/src/main/scala/org/apache/pekko/remote/testconductor/RemoteConnection.scala:
##########
@@ -56,8 +56,15 @@ private[pekko] class ProtobufEncoder extends MessageToMessageEncoder[Message] {
private[pekko] class ProtobufDecoder(prototype: Message) extends MessageToMessageDecoder[ByteBuf] {
override def decode(ctx: ChannelHandlerContext, msg: ByteBuf, out: java.util.List[AnyRef]): Unit = {
- val bytes = ByteBufUtil.getBytes(msg)
- out.add(prototype.getParserForType.parseFrom(bytes))
+ try {
+ val bytes = ByteBufUtil.getBytes(msg)
+ out.add(prototype.getParserForType.parseFrom(bytes))
+ } catch {
+ case NonFatal(e) =>
+ throw new RuntimeException("Error while decoding protobuf message", e)
Review Comment:
Can you suggest to Flink team to revert their upgrade to Pekko 1.1 everywhere if you think that this is definietly broken.
##########
multi-node-testkit/src/main/scala/org/apache/pekko/remote/testconductor/RemoteConnection.scala:
##########
@@ -56,8 +56,15 @@ private[pekko] class ProtobufEncoder extends MessageToMessageEncoder[Message] {
private[pekko] class ProtobufDecoder(prototype: Message) extends MessageToMessageDecoder[ByteBuf] {
override def decode(ctx: ChannelHandlerContext, msg: ByteBuf, out: java.util.List[AnyRef]): Unit = {
- val bytes = ByteBufUtil.getBytes(msg)
- out.add(prototype.getParserForType.parseFrom(bytes))
+ try {
+ val bytes = ByteBufUtil.getBytes(msg)
+ out.add(prototype.getParserForType.parseFrom(bytes))
+ } catch {
+ case NonFatal(e) =>
+ throw new RuntimeException("Error while decoding protobuf message", e)
Review Comment:
Can you suggest to Flink team to revert their upgrade to Pekko 1.1 everywhere if you think that this is definitely broken.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: Fix ByteBuf leak in remoting transport. [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1635:
URL: https://github.com/apache/pekko/pull/1635#discussion_r1898915568
##########
multi-node-testkit/src/main/scala/org/apache/pekko/remote/testconductor/RemoteConnection.scala:
##########
@@ -56,8 +56,15 @@ private[pekko] class ProtobufEncoder extends MessageToMessageEncoder[Message] {
private[pekko] class ProtobufDecoder(prototype: Message) extends MessageToMessageDecoder[ByteBuf] {
override def decode(ctx: ChannelHandlerContext, msg: ByteBuf, out: java.util.List[AnyRef]): Unit = {
- val bytes = ByteBufUtil.getBytes(msg)
- out.add(prototype.getParserForType.parseFrom(bytes))
+ try {
+ val bytes = ByteBufUtil.getBytes(msg)
+ out.add(prototype.getParserForType.parseFrom(bytes))
+ } catch {
+ case NonFatal(e) =>
+ throw new RuntimeException("Error while decoding protobuf message", e)
Review Comment:
OK, I will change it then .
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: Fix ByteBuf leak in remoting transport. [pekko]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning commented on PR #1635:
URL: https://github.com/apache/pekko/pull/1635#issuecomment-2564365975
I would prefer if the exception handling changes were tracked separately - separate issue with a description of why the existing exception handling is wrong - and separate PR.
I think this PR should just add the finally block with the buffer release. That is all that highlighted as wrong in the issue.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: Fix ByteBuf leak in remoting transport. [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin closed pull request #1635: fix: Fix ByteBuf leak in remoting transport.
URL: https://github.com/apache/pekko/pull/1635
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] fix: Fix ByteBuf leak in remoting transport. [pekko]
Posted by "He-Pin (via GitHub)" <gi...@apache.org>.
He-Pin commented on code in PR #1635:
URL: https://github.com/apache/pekko/pull/1635#discussion_r1898916610
##########
remote/src/main/scala/org/apache/pekko/remote/transport/netty/TcpSupport.scala:
##########
@@ -56,8 +58,16 @@ private[remote] trait TcpHandlers extends CommonHandlers {
}
override def onMessage(ctx: ChannelHandlerContext, msg: ByteBuf): Unit = {
- val bytes: Array[Byte] = ByteBufUtil.getBytes(msg)
- if (bytes.length > 0) notifyListener(ctx.channel(), InboundPayload(ByteString(bytes)))
+ try {
+ val bytes: Array[Byte] = ByteBufUtil.getBytes(msg)
+ if (bytes.length > 0) notifyListener(ctx.channel(), InboundPayload(ByteString(bytes)))
+ } catch {
+ case NonFatal(e) =>
+ log.warning("Error while handling message from [{}]: {}", ctx.channel().remoteAddress(), e)
+ onException(ctx, e)
+ } finally {
+ msg.release()
Review Comment:
Sorry for not checking the `LengthFieldBasedFrameDecoder` in detail, we need to turn on the leak detector in nightly testing I think.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org