[DRILL-6746] Query can hang when PartitionSender task thread sees a connection failure while sending data batches to remote fragment

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.13.0
• Fix Version/s: 1.15.0
• Component/s: Execution - Flow
• Labels: ready-to-commit
• Assignee: Sorabh Hamirwasia
• Reporter: Sorabh Hamirwasia
• Reviewer: Salim Achouche
• Created: 19/Sep/18 18:38
• Updated: 24/Sep/18 23:22
• Resolved: 24/Sep/18 23:22

Description

An UnorderedMuxExchange is implemented using UnorderedReceiver and HashPartitionSender. Muxer is used to improve the memory usage, such that when multiple minor fragments (let say n) running on a node is sending data to multiple other remote nodes minor fragments (let say m), then each sending fragment has to create m buffers for m receivers. In total on a single node that means creating mn buffers. Whereas with use of muxer what can be done is all the data from m minor fragment can be sent to 1 local minor fragment and that local minor fragment will instead send data to m remote fragments/nodes. Hence total buffer to use will be m only. 

There is a shared queue which is filled with RecordBatches by all the m sending minor fragments (which is received on Data client channel and ultimately populated by netty thread) and then it is consumed by local minor fragment which has PartitionSender—> UnorderedReceiver with each next() call. Hence the queue is filled and consumed by different thread. When PartitionSender receives an incoming batch then based on some heuristics it creates multiple PartitionTasks threads which all goes over this incoming batch and populates rows that falls in their range to their outgoing batch. The main local minor fragment thread waits until all task thread is completed or in an event of interrupt. After which it gets next() incoming batch. In this process once the output batch is full then it's sent to the remote nodes. All the sends are done asynchronously.

In this case while sending the outgoing batch by task thread if there is any failure then the executor state of the main local fragment thread (running partitionSender and Unordered receiver) is set to FAILED state asynchronously. Meanwhile next() call is made to get new incoming batch. There is a race condition between the check of executor thread state with next() call and when the FAILED state is set. Hence next() can be called before state is actually updated. With this next() call if there is no RecordBatch present in the queue then the main local fragment thread will call take() on buffer queue and will wait until it get's a new batch. Meanwhile the executor state might get updated and the the netty thread which receives the batch and tries to enqueue it in queue will see the updated state and release the received batch without putting it in shared queue. Since no new batch will be stored in shared queue going forward the main local minor fragment thread will be stuck forever unless a cancellation is explicitly done which will interrupt the stuck thread. This can result in query hang.

Logs for above investigation:

It looks like the intermediate fragment 2:3 started executing when it receives the record batch.

drill-cluster/drillbit.1.log:2018-08-08 00:27:34,210 [249580bc-5bca-b166-e906-084b35ecf30d:frag:2:3] INFO  o.a.d.e.w.fragment.FragmentExecutor - 249580bc-5bca-b166-e906-084b35ecf30d:2:3: State change requested AWAITING_ALLOCATION --> RUNNING

drill-cluster/drillbit.1.log:2018-08-08 00:27:34,210 [249580bc-5bca-b166-e906-084b35ecf30d:frag:2:3] INFO  o.a.d.e.w.f.FragmentStatusReporter - 249580bc-5bca-b166-e906-084b35ecf30d:2:3: State to report: RUNNING

But later while sending a record batch downstream to a remote node it receives failure as Data Connection is not established since remote Drillbit was not running then.

2018-08-08 00:33:29,184 [BitClient-7] ERROR o.a.d.e.rpc.ConnectionMultiListener - Failed to establish connection
java.util.concurrent.ExecutionException: io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: drillbit2/10.10.10.10:31012
        at io.netty.util.concurrent.AbstractFuture.get(AbstractFuture.java:54) ~[netty-common-4.0.48.Final.jar:4.0.48.Final]
        at org.apache.drill.exec.rpc.ConnectionMultiListener$ConnectionHandler.operationComplete(ConnectionMultiListener.java:90) [drill-rpc-1.13.0.jar:1.13.0]
        at org.apache.drill.exec.rpc.ConnectionMultiListener$ConnectionHandler.operationComplete(ConnectionMultiListener.java:77) [drill-rpc-1.13.0.jar:1.13.0]
        at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:507) [netty-common-4.0.48.Final.jar:4.0.48.Final]
        at io.netty.util.concurrent.DefaultPromise.notifyListeners0(DefaultPromise.java:500) [netty-common-4.0.48.Final.jar:4.0.48.Final]
        at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:479) [netty-common-4.0.48.Final.jar:4.0.48.Final]
        at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:420) [netty-common-4.0.48.Final.jar:4.0.48.Final]
        at io.netty.util.concurrent.DefaultPromise.tryFailure(DefaultPromise.java:122) [netty-common-4.0.48.Final.jar:4.0.48.Final]
        at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.fulfillConnectPromise(AbstractNioChannel.java:278) [netty-transport-4.0.48.Final.jar:4.0.48.Final]
        at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:294) [netty-transport-4.0.48.Final.jar:4.0.48.Final]
        at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:633) [netty-transport-4.0.48.Final.jar:4.0.48.Final]
        at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:580) [netty-transport-4.0.48.Final.jar:4.0.48.Final]
        at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:497) [netty-transport-4.0.48.Final.jar:4.0.48.Final]
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:459) [netty-transport-4.0.48.Final.jar:4.0.48.Final]
        at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:131) [netty-common-4.0.48.Final.jar:4.0.48.Final]
        at java.lang.Thread.run(Thread.java:748) [na:1.8.0_144]
Caused by: io.netty.channel.AbstractChannel$AnnotatedConnectException: Connection refused: drillbit2/10.10.10.10:31012
        at sun.nio.ch.SocketChannelImpl.checkConnect(Native Method) ~[na:1.8.0_144]
        at sun.nio.ch.SocketChannelImpl.finishConnect(SocketChannelImpl.java:717) ~[na:1.8.0_144]
        at io.netty.channel.socket.nio.NioSocketChannel.doFinishConnect(NioSocketChannel.java:258) ~[netty-transport-4.0.48.Final.jar:4.0.48.Final]
        at io.netty.channel.nio.AbstractNioChannel$AbstractNioUnsafe.finishConnect(AbstractNioChannel.java:291) [netty-transport-4.0.48.Final.jar:4.0.48.Final]
        ... 6 common frames omitted
Caused by: java.net.ConnectException: Connection refused
        ... 10 common frames omitted

Then due to this failure while sendingRecordBatch, the status handler set’s the Fragment Executor state to FAILED but cleanup is not performed as that will be done when main next() loop of operators is terminated.

2018-08-08 00:33:29,184 [BitClient-7] INFO  o.a.d.e.w.fragment.FragmentExecutor - 249580bc-5bca-b166-e906-084b35ecf30d:2:3: State change requested RUNNING --> FAILED

Ideally the operators should have seen the executor state is FAILED and should have started the cleanup process of FragmentExecutor. During cleanup the FragmentExecutor removes itself from runningFragments list and also removes its FragmentManager (being an intermediate Fragment). But looks like somewhere the executor thread was stuck and cleanup was not completed. The reason is because after 8 hours when cancellation was requested the Fragment Manager for this minor fragment was still found on that node and cancellation was performed properly.

2018-08-08 08:51:23,207 [BitServer-1] INFO  o.a.d.e.w.fragment.FragmentExecutor - 249580bc-5bca-b166-e906-084b35ecf30d:2:3: State change requested FAILED --> CANCELLATION_REQUESTED
2018-08-08 08:51:23,207 [BitServer-1] WARN  o.a.d.e.w.fragment.FragmentExecutor - 249580bc-5bca-b166-e906-084b35ecf30d:2:3: Ignoring unexpected state transition FAILED --> CANCELLATION_REQUESTED
2018-08-08 08:51:23,209 [249580bc-5bca-b166-e906-084b35ecf30d:frag:2:3] INFO  o.a.d.e.w.fragment.FragmentExecutor - 249580bc-5bca-b166-e906-084b35ecf30d:2:3: State change requested FAILED --> FINISHED

2018-08-08 08:51:23,216 [249580bc-5bca-b166-e906-084b35ecf30d:frag:2:3] ERROR o.a.d.e.w.fragment.FragmentExecutor - SYSTEM ERROR: ConnectException: Connection refused

Fragment 2:3

[Error Id: 3f66fbfb-a93b-4b48-8008-ebb78e084905 on drillbit1:31010]
org.apache.drill.common.exceptions.UserException: SYSTEM ERROR: ConnectException: Connection refused

Fragment 2:3

[Error Id: 3f66fbfb-a93b-4b48-8008-ebb78e084905 on drillbit1:31010]
        at org.apache.drill.common.exceptions.UserException$Builder.build(UserException.java:633) ~[drill-common-1.13.0.jar:1.13.0]
        at org.apache.drill.exec.work.fragment.FragmentExecutor.sendFinalState(FragmentExecutor.java:300) [drill-java-exec-1.13.0.jar:1.13.0]
        at org.apache.drill.exec.work.fragment.FragmentExecutor.cleanup(FragmentExecutor.java:160) [drill-java-exec-1.13.0.jar:1.13.0]
        at org.apache.drill.exec.work.fragment.FragmentExecutor.run(FragmentExecutor.java:266) [drill-java-exec-1.13.0.jar:1.13.0]
        at org.apache.drill.common.SelfCleaningRunnable.run(SelfCleaningRunnable.java:38) [drill-common-1.13.0.jar:1.13.0]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [na:1.8.0_144]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [na:1.8.0_144]
        at java.lang.Thread.run(Thread.java:748) [na:1.8.0_144]
Caused by: org.apache.drill.exec.rpc.RpcException: Command failed while establishing connection.  Failure type CONNECTION.
        at org.apache.drill.exec.rpc.RpcException.mapException(RpcException.java:67) ~[drill-rpc-1.13.0.jar:1.13.0]
        at org.apache.drill.exec.rpc.ListeningCommand.connectionFailed(ListeningCommand.java:66) ~[drill-rpc-1.13.0.jar:1.13.0]
        at org.apache.drill.exec.rpc.data.DataTunnel$SendBatchAsyncListen.connectionFailed(DataTunnel.java:145) ~[drill-java-exec-1.13.0.jar:1.13.0]
        at org.apache.drill.exec.rpc.ReconnectingConnection$ConnectionListeningFuture.connectionFailed(ReconnectingConnection.java:152) ~[drill-rpc-1.13.0.jar:1.13.0]
        at org.apache.drill.exec.rpc.ConnectionMultiListener$ConnectionHandler.operationComplete(ConnectionMultiListener.java:119) ~[drill-rpc-1.13.0.jar:1.13.0]
        at 
Caused by: java.net.ConnectException: Connection refused
        ... 10 common frames omitted

Cleanup can be stuck for 8 hours (before cancellation was triggered) because of below reasons.

1. This intermediate Fragment was running Unordered Receiver and PartitionSender. So if there is a logic in PartitionSender such that in some case it will keep waiting for next incoming batch without checking the fragment executor state using (shouldContinue()) then we can get stuck. Since only after checking shouldContinue() state it will know that it has to exit and not wait for any more batches. Still need to look into PartitionSender code and understand it.

Also I looked into profile and based on my understanding major fragment 2's unordered receiver should get all the batches from major fragment 4 single sender. When I add batches in single sender to receiver there is mismatch. Basically sender has sent more batches than what receiver has received:

Sender Stats:
Batches: 113,528
Row: 62,554,675

Receiver Stats:
Batches: 111,372
Row: 61,137,838

Issue Links

• links to: GitHub Pull Request #1470