diff options
| -rwxr-xr-x | build_integration.sh | 11 | ||||
| -rw-r--r-- | kafka/client_async.py | 14 | ||||
| -rw-r--r-- | kafka/conn.py | 10 | ||||
| -rw-r--r-- | kafka/socketpair.py | 57 | ||||
| -rw-r--r-- | test/test_consumer.py | 2 |
5 files changed, 83 insertions, 11 deletions
diff --git a/build_integration.sh b/build_integration.sh index 4785028..0babfa1 100755 --- a/build_integration.sh +++ b/build_integration.sh @@ -54,7 +54,16 @@ pushd servers fi if [ ! -f "../$kafka/kafka-bin/bin/kafka-run-class.sh" ]; then echo "Downloading kafka ${kafka} tarball" - wget -N https://archive.apache.org/dist/kafka/$kafka/${KAFKA_ARTIFACT}.tgz || wget -N https://archive.apache.org/dist/kafka/$kafka/${KAFKA_ARTIFACT}.tar.gz + if hash wget 2>/dev/null; then + wget -N https://archive.apache.org/dist/kafka/$kafka/${KAFKA_ARTIFACT}.tgz || wget -N https://archive.apache.org/dist/kafka/$kafka/${KAFKA_ARTIFACT}.tar.gz + else + echo "wget not found... using curl" + if [ -f "${KAFKA_ARTIFACT}.tar.gz" ]; then + echo "Using cached artifact: ${KAFKA_ARTIFACT}.tar.gz" + else + curl -f https://archive.apache.org/dist/kafka/$kafka/${KAFKA_ARTIFACT}.tgz -o ${KAFKA_ARTIFACT}.tar.gz || curl -f https://archive.apache.org/dist/kafka/$kafka/${KAFKA_ARTIFACT}.tar.gz -o ${KAFKA_ARTIFACT}.tar.gz + fi + fi echo echo "Extracting kafka ${kafka} binaries" tar xzvf ${KAFKA_ARTIFACT}.t* -C ../$kafka/ diff --git a/kafka/client_async.py b/kafka/client_async.py index 57aea66..ae9dbb4 100644 --- a/kafka/client_async.py +++ b/kafka/client_async.py @@ -4,9 +4,9 @@ import copy import heapq import itertools import logging -import os import random import select +import socket import time import six @@ -18,6 +18,7 @@ from .conn import BrokerConnection, ConnectionStates, collect_hosts from .future import Future from .protocol.metadata import MetadataRequest from .protocol.produce import ProduceRequest +from . import socketpair from .version import __version__ if six.PY2: @@ -97,11 +98,11 @@ class KafkaClient(object): self._last_bootstrap = 0 self._bootstrap_fails = 0 self._bootstrap(collect_hosts(self.config['bootstrap_servers'])) - self._wake_r, self._wake_w = os.pipe() + self._wake_r, self._wake_w = socket.socketpair() def __del__(self): - os.close(self._wake_r) - os.close(self._wake_w) + self._wake_r.close() + self._wake_w.close() def _bootstrap(self, hosts): # Exponential backoff if bootstrap fails @@ -674,14 +675,15 @@ class KafkaClient(object): return version def wakeup(self): - os.write(self._wake_w, b'x') + if self._wake_w.send(b'x') != 1: + log.warning('Unable to send to wakeup socket!') def _clear_wake_fd(self): while True: fds, _, _ = select.select([self._wake_r], [], [], 0) if not fds: break - os.read(self._wake_r, 1) + self._wake_r.recv(1) class DelayedTaskQueue(object): diff --git a/kafka/conn.py b/kafka/conn.py index 015bf23..65451f9 100644 --- a/kafka/conn.py +++ b/kafka/conn.py @@ -90,9 +90,10 @@ class BrokerConnection(object): pass self.last_attempt = time.time() - if not ret or ret is errno.EISCONN: + if not ret or ret == errno.EISCONN: self.state = ConnectionStates.CONNECTED - elif ret in (errno.EINPROGRESS, errno.EALREADY): + # WSAEINVAL == 10022, but errno.WSAEINVAL is not available on non-win systems + elif ret in (errno.EINPROGRESS, errno.EALREADY, errno.EWOULDBLOCK, 10022): self.state = ConnectionStates.CONNECTING else: log.error('Connect attempt to %s returned error %s.' @@ -114,9 +115,10 @@ class BrokerConnection(object): ret = self._sock.connect_ex((self.host, self.port)) except socket.error as ret: pass - if not ret or ret is errno.EISCONN: + if not ret or ret == errno.EISCONN: self.state = ConnectionStates.CONNECTED - elif ret is not errno.EALREADY: + # WSAEINVAL == 10022, but errno.WSAEINVAL is not available on non-win systems + elif ret not in (errno.EINPROGRESS, errno.EALREADY, errno.EWOULDBLOCK, 10022): log.error('Connect attempt to %s returned error %s.' ' Disconnecting.', self, ret) self.close() diff --git a/kafka/socketpair.py b/kafka/socketpair.py new file mode 100644 index 0000000..6a87c4d --- /dev/null +++ b/kafka/socketpair.py @@ -0,0 +1,57 @@ +# pylint: skip-file +# vendored from https://github.com/mhils/backports.socketpair + +import sys +import socket +import errno + +_LOCALHOST = '127.0.0.1' +_LOCALHOST_V6 = '::1' + +if not hasattr(socket, "socketpair"): + # Origin: https://gist.github.com/4325783, by Geert Jansen. Public domain. + def socketpair(family=socket.AF_INET, type=socket.SOCK_STREAM, proto=0): + if family == socket.AF_INET: + host = _LOCALHOST + elif family == socket.AF_INET6: + host = _LOCALHOST_V6 + else: + raise ValueError("Only AF_INET and AF_INET6 socket address families " + "are supported") + if type != socket.SOCK_STREAM: + raise ValueError("Only SOCK_STREAM socket type is supported") + if proto != 0: + raise ValueError("Only protocol zero is supported") + + # We create a connected TCP socket. Note the trick with + # setblocking(False) that prevents us from having to create a thread. + lsock = socket.socket(family, type, proto) + try: + lsock.bind((host, 0)) + lsock.listen(min(socket.SOMAXCONN, 128)) + # On IPv6, ignore flow_info and scope_id + addr, port = lsock.getsockname()[:2] + csock = socket.socket(family, type, proto) + try: + csock.setblocking(False) + if sys.version_info >= (3, 0): + try: + csock.connect((addr, port)) + except (BlockingIOError, InterruptedError): + pass + else: + try: + csock.connect((addr, port)) + except socket.error as e: + if e.errno != errno.WSAEWOULDBLOCK: + raise + csock.setblocking(True) + ssock, _ = lsock.accept() + except: + csock.close() + raise + finally: + lsock.close() + return (ssock, csock) + + socket.socketpair = socketpair diff --git a/test/test_consumer.py b/test/test_consumer.py index 2c9561b..e664292 100644 --- a/test/test_consumer.py +++ b/test/test_consumer.py @@ -1,3 +1,4 @@ +import sys from mock import MagicMock, patch from . import unittest @@ -17,6 +18,7 @@ class TestKafkaConsumer(unittest.TestCase): class TestMultiProcessConsumer(unittest.TestCase): + @unittest.skipIf(sys.platform.startswith('win'), 'test mocking fails on windows') def test_partition_list(self): client = MagicMock() partitions = (0,) |