Fix bug in socket timeout per PR #161 by maciejkula, add test

kafka/conn.py

@@ -150,6 +150,6 @@ class KafkaConnection(local):
         """
         self.close()
         self._sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        self._sock.connect((self.host, self.port))
         self._sock.settimeout(self.timeout)
+        self._sock.connect((self.host, self.port))
         self._dirty = False

test/test_client_integration.py

@@ -1,9 +1,10 @@
 import unittest
 import time
+import socket
+import random
 
-from kafka import *  # noqa
+import kafka
-from kafka.common import *  # noqa
+from kafka.common import *
-from kafka.codec import has_gzip, has_snappy
 from .fixtures import ZookeeperFixture, KafkaFixture
 from .testutil import *
 
@@ -19,6 +20,15 @@ class TestKafkaClientIntegration(KafkaIntegrationTestCase):
         cls.server.close()
         cls.zk.close()
 
+    def test_timeout(self):
+        server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        server_socket.bind(('localhost', 14567))
+
+        with Timer() as t:
+            with self.assertRaises((socket.timeout, socket.error)):
+                conn = kafka.conn.KafkaConnection("localhost", 14567, 1.0)
+        self.assertGreaterEqual(t.interval, 1.0)
+
     def test_consume_none(self):
         fetch = FetchRequest(self.topic, 0, 0, 1024)