[CAMEL-8193] Frequent BlockingOperationExceptions under load

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 2.14.1
• Fix Version/s: 2.14.2, 2.15.0
• Component/s: camel-netty, camel-netty4

Environment

Ubuntu 14.04, JDK 1.8.0_25

• Patch Info: Patch Available
• Estimated Complexity: Unknown
• Assignee: Willem Jiang
• Reporter: Mark Mindenhall
• Created: 30/Dec/14 06:19
• Updated: 14/Jan/15 10:53
• Resolved: 14/Jan/15 10:53

Description

I have a class that combines a netty4 consumer with a kafka producer, pulling messages off of a TCP socket and sticking them into kafka for further processing. While doing some load testing, I started seeing a lot of stack traces like the following:

2014-12-29 19:55:57,517 - [WARN] - from io.netty.util.concurrent.DefaultPromise in Camel Thread #23 - NettyServerTCPWorker
An exception was thrown by org.apache.camel.component.netty4.handlers.ServerResponseFutureListener.operationComplete()
io.netty.util.concurrent.BlockingOperationException: DefaultChannelPromise@4e2727d0(uncancellable)
	at io.netty.util.concurrent.DefaultPromise.checkDeadLock(DefaultPromise.java:390) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.channel.DefaultChannelPromise.checkDeadLock(DefaultChannelPromise.java:157) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.util.concurrent.DefaultPromise.awaitUninterruptibly(DefaultPromise.java:283) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.channel.DefaultChannelPromise.awaitUninterruptibly(DefaultChannelPromise.java:135) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.channel.DefaultChannelPromise.awaitUninterruptibly(DefaultChannelPromise.java:28) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.util.concurrent.DefaultPromise.syncUninterruptibly(DefaultPromise.java:225) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.channel.DefaultChannelPromise.syncUninterruptibly(DefaultChannelPromise.java:123) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.channel.DefaultChannelPromise.syncUninterruptibly(DefaultChannelPromise.java:28) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at org.apache.camel.component.netty4.NettyHelper.close(NettyHelper.java:121) ~[camel-netty4-2.14.1.jar:2.14.1]
	at org.apache.camel.component.netty4.handlers.ServerResponseFutureListener.operationComplete(ServerResponseFutureListener.java:80) ~[camel-netty4-2.14.1.jar:2.14.1]
	at org.apache.camel.component.netty4.handlers.ServerResponseFutureListener.operationComplete(ServerResponseFutureListener.java:36) ~[camel-netty4-2.14.1.jar:2.14.1]
	at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:680) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:567) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.util.concurrent.DefaultPromise.trySuccess(DefaultPromise.java:406) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.channel.DefaultChannelPromise.trySuccess(DefaultChannelPromise.java:82) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.channel.ChannelOutboundBuffer.safeSuccess(ChannelOutboundBuffer.java:523) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.channel.ChannelOutboundBuffer.remove(ChannelOutboundBuffer.java:247) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.channel.ChannelOutboundBuffer.removeBytes(ChannelOutboundBuffer.java:322) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.channel.socket.nio.NioSocketChannel.doWrite(NioSocketChannel.java:301) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.channel.AbstractChannel$AbstractUnsafe.flush0(AbstractChannel.java:745) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.flush0(AbstractNioChannel.java:315) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.channel.AbstractChannel$AbstractUnsafe.flush(AbstractChannel.java:713) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.channel.DefaultChannelPipeline$HeadContext.flush(DefaultChannelPipeline.java:1159) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.channel.ChannelHandlerInvokerUtil.invokeFlushNow(ChannelHandlerInvokerUtil.java:165) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.channel.DefaultChannelHandlerInvoker.invokeFlush(DefaultChannelHandlerInvoker.java:355) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.channel.AbstractChannelHandlerContext.flush(AbstractChannelHandlerContext.java:297) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.channel.ChannelOutboundHandlerAdapter.flush(ChannelOutboundHandlerAdapter.java:115) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.channel.ChannelHandlerInvokerUtil.invokeFlushNow(ChannelHandlerInvokerUtil.java:165) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.channel.DefaultChannelHandlerInvoker$16.run(DefaultChannelHandlerInvoker.java:363) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:343) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:356) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:834) [netty-all-4.1.0.Beta3.jar:4.1.0.Beta3]
	at java.lang.Thread.run(Thread.java:745) [na:1.8.0_20]

The documentation for ChannelFuture says "Do not call await() inside ChannelHandler", due to a possible deadlock. Per the documentation, this results in an IllegalStateException (netty3) or a BlockingOperationException (netty4+).

The javadoc for NettyHelper.close claims to close the channel asynchronously, but that method calls channel.close().syncUninterruptibly(), which blocks waiting for the channel to close and results in the exception under certain conditions.

The attached patch resolves this issue for both the camel-netty and camel-netty4 components.

Issue Links

• is related to: CAMEL-7909 camel-netty-http consumer need to close the connection if the response connection header is close