Improve Zookeeper / Kafka Fixture management

  - spawn fixtures via daemon threads
  - close fixtures atexit and in __del__ to avoid interpreter hangs
  - raise Exception on timeouts in open()

1 files changed, 21 insertions, 4 deletions

diff --git a/test/fixtures.py b/test/fixtures.py
index 3892416..3c5e694 100644
--- a/test/fixtures.py
+++ b/test/fixtures.py
@@ -1,3 +1,4 @@
+import atexit
 import logging
 import os
 import os.path
@@ -130,24 +131,33 @@ class ZookeeperFixture(Fixture):
         timeout = 5
         max_timeout = 30
         backoff = 1
-        while True:
+        end_at = time.time() + max_timeout
+        while time.time() < end_at:
             self.child = SpawnedService(args, env)
             self.child.start()
-            timeout = min(timeout, max_timeout)
+            timeout = min(timeout, max(end_at - time.time(), 0))
             if self.child.wait_for(r"binding to port", timeout=timeout):
                 break
             self.child.stop()
             timeout *= 2
             time.sleep(backoff)
+        else:
+            raise Exception('Failed to start Zookeeper before max_timeout')
         self.out("Done!")
+        atexit.register(self.close)
 
     def close(self):
+        if self.child is None:
+            return
         self.out("Stopping...")
         self.child.stop()
         self.child = None
         self.out("Done!")
         shutil.rmtree(self.tmp_dir)
 
+    def __del__(self):
+        self.close()
+
 
 class KafkaFixture(Fixture):
     @classmethod
@@ -240,18 +250,25 @@ class KafkaFixture(Fixture):
         timeout = 5
         max_timeout = 30
         backoff = 1
-        while True:
+        end_at = time.time() + max_timeout
+        while time.time() < end_at:
             self.child = SpawnedService(args, env)
             self.child.start()
-            timeout = min(timeout, max_timeout)
+            timeout = min(timeout, max(end_at - time.time(), 0))
             if self.child.wait_for(r"\[Kafka Server %d\], Started" %
                                    self.broker_id, timeout=timeout):
                 break
             self.child.stop()
             timeout *= 2
             time.sleep(backoff)
+        else:
+            raise Exception('Failed to start KafkaInstance before max_timeout')
         self.out("Done!")
         self.running = True
+        atexit.register(self.close)
+
+    def __del__(self):
+        self.close()
 
     def close(self):
         if not self.running: