rdkafka/producer: minor changes to poll()

Checking for ReferenceError is just noise since we always join() the
poller thread before stopping/finalising.

Signed-off-by: Yung-Chin Oei <[email protected]>

Author: yungchin

pykafka/rdkafka/producer.py

@@ -52,11 +52,9 @@ def start(self):
             self._running = True
 
             def poll(self):
-                try:
-                    while self._running or self._rdk_producer.outq_len() > 0:
-                        self._rdk_producer.poll(timeout_ms=1000)
-                except ReferenceError:  # weakref'd self
-                    pass
+                while self._running or self._rdk_producer.outq_len() > 0:
+                    self._rdk_producer.poll(timeout_ms=1000)
+                assert(not self._rdk_producer._pending_messages)
                 log.debug("Exiting RdKafkaProducer poller thread cleanly.")
 
             self._poller_thread = self._cluster.handler.spawn(