2
0
mirror of https://github.com/xcat2/confluent.git synced 2026-07-30 17:49:48 +00:00

Add a recourse to drain queue

While it may not have been possible in eventlet for this to happen,
strictly speaking if it were a thread, it could exit during check for
liveness and leave data on the queue.

To be careful, also drain the queue after all children dead.
This commit is contained in:
Jarrod Johnson
2017-04-25 16:48:35 -04:00
parent cf97bbe299
commit 6ab4d3c886
@@ -290,6 +290,15 @@ def perform_requests(operator, nodes, element, cfg, inputdata):
for t in list(livingthreads):
if t.dead:
livingthreads.discard(t)
try:
# drain queue if a thread put something on the queue and died
while True:
datum = resultdata.get_nowait()
if datum != 'Done':
yield datum
except queue.Empty:
pass
def perform_request(operator, node, element,