diff options
| -rw-r--r-- | kafka/client.py | 2 | ||||
| -rw-r--r-- | kafka/consumer/multiprocess.py | 15 | ||||
| -rw-r--r-- | kafka/consumer/simple.py | 27 |
3 files changed, 21 insertions, 23 deletions
diff --git a/kafka/client.py b/kafka/client.py index 63b33b3..63c9073 100644 --- a/kafka/client.py +++ b/kafka/client.py @@ -4,8 +4,8 @@ import copy import functools import logging import time -import kafka.common +import kafka.common from kafka.common import (TopicAndPartition, BrokerMetadata, ConnectionError, FailedPayloadsError, KafkaTimeoutError, KafkaUnavailableError, diff --git a/kafka/consumer/multiprocess.py b/kafka/consumer/multiprocess.py index 8cec92d..d03eb95 100644 --- a/kafka/consumer/multiprocess.py +++ b/kafka/consumer/multiprocess.py @@ -1,22 +1,21 @@ from __future__ import absolute_import -import logging -import time - from collections import namedtuple +import logging from multiprocessing import Process, Manager as MPManager - try: - from Queue import Empty, Full -except ImportError: # python 2 - from queue import Empty, Full + from Queue import Empty, Full # python 3 +except ImportError: + from queue import Empty, Full # python 2 +import time from .base import ( + Consumer, AUTO_COMMIT_MSG_COUNT, AUTO_COMMIT_INTERVAL, NO_MESSAGES_WAIT_TIME_SECONDS, FULL_QUEUE_WAIT_TIME_SECONDS ) -from .simple import Consumer, SimpleConsumer +from .simple import SimpleConsumer log = logging.getLogger(__name__) diff --git a/kafka/consumer/simple.py b/kafka/consumer/simple.py index 384fa8e..e4233ff 100644 --- a/kafka/consumer/simple.py +++ b/kafka/consumer/simple.py @@ -2,25 +2,18 @@ from __future__ import absolute_import try: from itertools import zip_longest as izip_longest, repeat # pylint: disable-msg=E0611 -except ImportError: # python 2 - from itertools import izip_longest as izip_longest, repeat +except ImportError: + from itertools import izip_longest as izip_longest, repeat # python 2 import logging +try: + from Queue import Empty, Queue # python 3 +except ImportError: + from queue import Empty, Queue # python 2 +import sys import time import six -import sys - -try: - from Queue import Empty, Queue -except ImportError: # python 2 - from queue import Empty, Queue -from kafka.common import ( - FetchRequest, OffsetRequest, - ConsumerFetchSizeTooSmall, ConsumerNoMoreData, - UnknownTopicOrPartitionError, NotLeaderForPartitionError, - OffsetOutOfRangeError, FailedPayloadsError, check_error -) from .base import ( Consumer, FETCH_DEFAULT_BLOCK_TIMEOUT, @@ -33,6 +26,12 @@ from .base import ( ITER_TIMEOUT_SECONDS, NO_MESSAGES_WAIT_TIME_SECONDS ) +from ..common import ( + FetchRequest, OffsetRequest, + ConsumerFetchSizeTooSmall, ConsumerNoMoreData, + UnknownTopicOrPartitionError, NotLeaderForPartitionError, + OffsetOutOfRangeError, FailedPayloadsError, check_error +) log = logging.getLogger(__name__) |