You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "poorbarcode (via GitHub)" <gi...@apache.org> on 2023/10/12 14:01:45 UTC
[PR] [fix] [client] Fix IO buffer overflow when resend msg after producer reconnect [pulsar]
poorbarcode opened a new pull request, #21351:
URL: https://github.com/apache/pulsar/pull/21351
### Motivation
If too many messages were called `sendAsync` when the state of the producer is reconnecting, after connection successfully, the messages cached in the memory will be flushed to the socket once, then the buffer overflow.
```
Got exception java.lang.IllegalStateException: buffer queue length overflow: 2143977581 + 4195072
at io.netty.channel.AbstractCoalescingBufferQueue.incrementReadableBytes(AbstractCoalescingBufferQueue.java:368)
at io.netty.channel.AbstractCoalescingBufferQueue.add(AbstractCoalescingBufferQueue.java:100)
at io.netty.channel.AbstractCoalescingBufferQueue.add(AbstractCoalescingBufferQueue.java:84)
at io.netty.handler.ssl.SslHandler.write(SslHandler.java:758)
at io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:881)
at io.netty.channel.AbstractChannelHandlerContext.invokeWrite(AbstractChannelHandlerContext.java:863)
at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:968)
at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:856)
at org.apache.pulsar.common.protocol.ByteBufPair$CopyingEncoder.write(ByteBufPair.java:149)
at io.netty.channel.AbstractChannelHandlerContext.invokeWrite0(AbstractChannelHandlerContext.java:881)
at io.netty.channel.AbstractChannelHandlerContext.invokeWrite(AbstractChannelHandlerContext.java:863)
at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:968)
at io.netty.channel.AbstractChannelHandlerContext.write(AbstractChannelHandlerContext.java:856)
at org.apache.pulsar.client.impl.ProducerImpl.recoverProcessOpSendMsgFrom(ProducerImpl.java:2283)
at org.apache.pulsar.client.impl.ProducerImpl.lambda$resendMessages$17(ProducerImpl.java:1918)
at io.netty.util.concurrent.AbstractEventExecutor.runTask(AbstractEventExecutor.java:174)
at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java:167)
at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:470)
at io.netty.channel.epoll.EpollEventLoop.run(EpollEventLoop.java:403)
at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
at java.base/java.lang.Thread.run(Thread.java:833)
```
### Modifications
- To avoid IO buffer overflow, split to multi-flush.
- TODO add a test
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
### Matching PR in forked repository
PR in forked repository: x
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [drafr] [fix] [client] Fix IO buffer overflow when resend msg after producer reconnect [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat closed pull request #21351: [drafr] [fix] [client] Fix IO buffer overflow when resend msg after producer reconnect
URL: https://github.com/apache/pulsar/pull/21351
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [drafr] [fix] [client] Fix IO buffer overflow when resend msg after producer reconnect [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat commented on PR #21351:
URL: https://github.com/apache/pulsar/pull/21351#issuecomment-1774174043
I'm still not sure how it gets to 2GB of accumulated data. The client memory limit backpressure should have blocked long before the 2GB.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [drafr] [fix] [client] Fix IO buffer overflow when resend msg after producer reconnect [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #21351:
URL: https://github.com/apache/pulsar/pull/21351#discussion_r1357415575
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -2284,12 +2285,18 @@ private void recoverProcessOpSendMsgFrom(ClientCnx cnx, MessageImpl from, long e
if (stripChecksum) {
stripChecksum(op);
}
+ // To avoid IO buffer overflow, split to multi-flush.
+ if (messageBytesSizeInCache + op.cmd.readableBytes() < messageBytesSizeInCache) {
+ cnx.ctx().flush();
Review Comment:
`flush` is an async operation, so this might not work as expected
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [drafr] [fix] [client] Fix IO buffer overflow when resend msg after producer reconnect [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #21351:
URL: https://github.com/apache/pulsar/pull/21351#discussion_r1359204213
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -2284,12 +2285,18 @@ private void recoverProcessOpSendMsgFrom(ClientCnx cnx, MessageImpl from, long e
if (stripChecksum) {
stripChecksum(op);
}
+ // To avoid IO buffer overflow, split to multi-flush.
+ if (messageBytesSizeInCache + op.cmd.readableBytes() < messageBytesSizeInCache) {
+ cnx.ctx().flush();
Review Comment:
I mean from a different aspect. I guess that in this case, the purpose of flushing is to ensure that buffers don't overflow. It feels like the solution in this PR won't fully address that since all operations will be queued up in any case unless the logic that calls write is also run in the ctx loop (thread). Other possibility would be to somehow wait for flush completion until it finishes.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [drafr] [fix] [client] Fix IO buffer overflow when resend msg after producer reconnect [pulsar]
Posted by "poorbarcode (via GitHub)" <gi...@apache.org>.
poorbarcode commented on code in PR #21351:
URL: https://github.com/apache/pulsar/pull/21351#discussion_r1359155689
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -2284,12 +2285,18 @@ private void recoverProcessOpSendMsgFrom(ClientCnx cnx, MessageImpl from, long e
if (stripChecksum) {
stripChecksum(op);
}
+ // To avoid IO buffer overflow, split to multi-flush.
+ if (messageBytesSizeInCache + op.cmd.readableBytes() < messageBytesSizeInCache) {
+ cnx.ctx().flush();
Review Comment:
Since `write` and `flush` will be executed in the same thread, the actions of both `write` and `flush` will keep the order to execute. So it will work as expected.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [drafr] [fix] [client] Fix IO buffer overflow when resend msg after producer reconnect [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #21351:
URL: https://github.com/apache/pulsar/pull/21351#discussion_r1362396001
##########
pulsar-client/src/main/java/org/apache/pulsar/client/impl/ProducerImpl.java:
##########
@@ -2284,12 +2285,18 @@ private void recoverProcessOpSendMsgFrom(ClientCnx cnx, MessageImpl from, long e
if (stripChecksum) {
stripChecksum(op);
}
+ // To avoid IO buffer overflow, split to multi-flush.
+ if (messageBytesSizeInCache + op.cmd.readableBytes() < messageBytesSizeInCache) {
+ cnx.ctx().flush();
Review Comment:
In Netty, there's the `Channel.isWritable` method and [`channelWritabilityChanged`](https://netty.io/4.1/api/io/netty/channel/ChannelInboundHandler.html#channelWritabilityChanged-io.netty.channel.ChannelHandlerContext-) callback that help in keeping the queued writes bounded (see [low/high watermark options](https://netty.io/4.0/api/io/netty/channel/ChannelConfig.html#setWriteBufferHighWaterMark-int-)). However, IIRC, Pulsar code base doesn't show examples of how write logic could be implemented to take advantage of this type of backpressure solution.
Optimally, the logic would be implemented in a way where more writes are added while the channel is writable and then paused. Adding more writes should resume after the channel becomes writable again.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [drafr] [fix] [client] Fix IO buffer overflow when resend msg after producer reconnect [pulsar]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat commented on PR #21351:
URL: https://github.com/apache/pulsar/pull/21351#issuecomment-6353892599
Thanks @poorbarcode. Re-checked against master: the overflow needs >2 GiB pending on one TLS connection, which the client memory limit (64 MiB default) prevents unless it's raised past 2 GiB or disabled. The intermediate `flush()` here also wouldn't drain the SSL queue: `FlushConsolidationHandler` defers it to a later task. A real fix would pace the resend on `channel.isWritable()`/`channelWritabilityChanged`, as @lhotari suggested. Closing for now; please reopen with a repro if you hit this with a bounded memory limit.
--
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: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org