[beam] 02/04: Adding msecs tracking to side inputs iterator

[pabloem]

This is an automated email from the ASF dual-hosted git repository.

pabloem pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git

commit 4f7c68d7ca562861e563f1119396140fec40e936
Author: Pablo <pabl...@google.com>
AuthorDate: Tue May 8 12:08:37 2018 -0700

    Adding msecs tracking to side inputs iterator
---
 sdks/python/apache_beam/runners/worker/sideinputs.py | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/sdks/python/apache_beam/runners/worker/sideinputs.py 
b/sdks/python/apache_beam/runners/worker/sideinputs.py
index c36182d..a71557c 100644
--- a/sdks/python/apache_beam/runners/worker/sideinputs.py
+++ b/sdks/python/apache_beam/runners/worker/sideinputs.py
@@ -159,7 +159,8 @@ class PrefetchingSourceSetIterable(object):
     try:
       while True:
         try:
-          element = self.element_queue.get()
+          with self.read_counter:
+            element = self.element_queue.get()
           if element is READER_THREAD_IS_DONE_SENTINEL:
             num_readers_finished += 1
             if num_readers_finished == self.num_reader_threads:

-- 
To stop receiving notification emails like this one, please contact
pabl...@apache.org.