Fix python3 / python2 comments re queue/Queue

Author: Dana Powers

kafka/consumer/multiprocess.py

@@ -4,9 +4,9 @@
 import logging
 from multiprocessing import Process, Manager as MPManager
 try:
-    from Queue import Empty, Full # python 3
+    import queue # python 3
 except ImportError:
-    from queue import Empty, Full # python 2
+    import Queue as queue # python 2
 import time
 
 from ..common import KafkaError
@@ -71,7 +71,7 @@ def _mp_consume(client, group, topic, queue, size, events, **consumer_options):
                         try:
                             queue.put(message, timeout=FULL_QUEUE_WAIT_TIME_SECONDS)
                             break
-                        except Full:
+                        except queue.Full:
                             if events.exit.is_set(): break
 
                     count += 1
@@ -220,7 +220,7 @@ def __iter__(self):
                 # TODO: This is a hack and will make the consumer block for
                 # at least one second. Need to find a better way of doing this
                 partition, message = self.queue.get(block=True, timeout=1)
-            except Empty:
+            except queue.Empty:
                 break
 
             # Count, check and commit messages if necessary
@@ -270,7 +270,7 @@ def get_messages(self, count=1, block=True, timeout=10):
             try:
                 partition, message = self.queue.get(block_next_call,
                                                     timeout)
-            except Empty:
+            except queue.Empty:
                 break
 
             _msg = (partition, message) if self.partition_info else message

kafka/consumer/simple.py

@@ -6,9 +6,9 @@
     from itertools import izip_longest as izip_longest, repeat # python 2
 import logging
 try:
-    from Queue import Empty, Queue # python 3
+    import queue # python 3
 except ImportError:
-    from queue import Empty, Queue # python 2
+    import Queue as queue # python 2
 import sys
 import time
 
@@ -136,7 +136,7 @@ def __init__(self, client, group, topic, auto_commit=True, partitions=None,
         self.fetch_offsets = self.offsets.copy()
         self.iter_timeout = iter_timeout
         self.auto_offset_reset = auto_offset_reset
-        self.queue = Queue()
+        self.queue = queue.Queue()
 
     def __repr__(self):
         return '<SimpleConsumer group=%s, topic=%s, partitions=%s>' % \
@@ -257,7 +257,7 @@ def seek(self, offset, whence=None, partition=None):
         if self.auto_commit:
             self.commit()
 
-        self.queue = Queue()
+        self.queue = queue.Queue()
 
     def get_messages(self, count=1, block=True, timeout=0.1):
         """
@@ -341,7 +341,7 @@ def _get_message(self, block=True, timeout=0.1, get_partition_info=None,
                 return partition, message
             else:
                 return message
-        except Empty:
+        except queue.Empty:
             log.debug('internal queue empty after fetch - returning None')
             return None