Re-init the sockets in the new process

3 files changed, 16 insertions, 1 deletions

diff --git a/kafka/client.py b/kafka/client.py
index a1c2133..525551e 100644
--- a/kafka/client.py
+++ b/kafka/client.py
@@ -176,6 +176,10 @@ class KafkaClient(object):
         for conn in self.conns.values():
             conn.close()
 
+    def reinit(self):
+        for conn in self.conns.values():
+            conn.reinit()
+
     def send_produce_request(self, payloads=[], acks=1, timeout=1000,
                              fail_on_error=True, callback=None):
         """
diff --git a/kafka/conn.py b/kafka/conn.py
index fce1fdc..01975e4 100644
--- a/kafka/conn.py
+++ b/kafka/conn.py
@@ -86,3 +86,12 @@ class KafkaConnection(local):
     def close(self):
         "Close this connection"
         self._sock.close()
+
+    def reinit(self):
+        """
+        Re-initialize the socket connection
+        """
+        self._sock.close()
+        self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self._sock.connect((self.host, self.port))
+        self._sock.settimeout(10)
diff --git a/kafka/consumer.py b/kafka/consumer.py
index 67e1b10..5ca90de 100644
--- a/kafka/consumer.py
+++ b/kafka/consumer.py
@@ -445,7 +445,6 @@ class MultiProcessConsumer(Consumer):
             proc = Process(target=self._consume, args=(chunk,))
             proc.daemon = True
             proc.start()
-            time.sleep(0.2)
             self.procs.append(proc)
 
     def _consume(self, partitions):
@@ -454,6 +453,9 @@ class MultiProcessConsumer(Consumer):
         notifications given by the controller process
         """
 
+        # Make the child processes open separate socket connections
+        self.client.reinit()
+
         # We will start consumers without auto-commit. Auto-commit will be
         # done by the master controller process.
         consumer = SimpleConsumer(self.client, self.group, self.topic,