[
https://issues.apache.org/jira/browse/DRILL-6746?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16624147#comment-16624147
]
ASF GitHub Bot commented on DRILL-6746:
---------------------------------------
sachouche commented on a change in pull request #1470: DRILL-6746: Query can
hang when PartitionSender task thread sees a co…
URL: https://github.com/apache/drill/pull/1470#discussion_r219618697
##########
File path:
exec/java-exec/src/main/java/org/apache/drill/exec/work/batch/UnlimitedRawBatchBuffer.java
##########
@@ -63,6 +64,15 @@ public RawFragmentBatch take() throws IOException,
InterruptedException {
return batch;
}
+ @Override
+ public RawFragmentBatch poll(long timeout, TimeUnit timeUnit) throws
InterruptedException, IOException {
+ RawFragmentBatch batch = buffer.poll(timeout, timeUnit);
Review comment:
Does poll return null when timeout is reached? if not, then you should
handle timeout exception.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
> Query can hang when PartitionSender task thread sees a connection failure
> while sending data batches to remote fragment
> -----------------------------------------------------------------------------------------------------------------------
>
> Key: DRILL-6746
> URL: https://issues.apache.org/jira/browse/DRILL-6746
> Project: Apache Drill
> Issue Type: Bug
> Components: Execution - Flow
> Affects Versions: 1.13.0
> Reporter: Sorabh Hamirwasia
> Assignee: Sorabh Hamirwasia
> Priority: Major
> Fix For: 1.15.0
>
>
> 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.
> {code:java}
> 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
> {code}
> 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.
> {code:java}
> 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
> {code}
> 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.
> {code:java}
> 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
> {code}
> 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.
> {code:java}
> 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
> {code}
> Cleanup can be stuck for 8 hours (before cancellation was triggered) because
> of below reasons.
> # 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
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
