Skip to content

A simpler kafka consumer #234

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 33 commits into from
Dec 17, 2014
Merged

A simpler kafka consumer #234

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
33 commits
Select commit Hold shift + click to select a range
fa6738b
A simpler kafka consumer:
Sep 11, 2014
82b3e01
Handle FailedPayloadsError on client.send_fetch_request; permit offse…
Sep 11, 2014
b264d8f
add private methods _set_consumer_timeout_start() and _check_consumer…
Sep 11, 2014
5376351
Use configure() to check and set configuration keys
Sep 14, 2014
4c111d3
Use client.get_partition_ids_for_topic
Sep 14, 2014
62a7189
Add set_topic_partitions method to configure topics/partitions to con…
Sep 14, 2014
e702880
self._topics is private; fixup topic iterations for new TopicAndParti…
Sep 14, 2014
80dfaeb
Move kafka._msg_iter initialization from __init__() to next()
Sep 15, 2014
93dd705
_client is private var
Sep 15, 2014
d5ec0a6
Support setting offsets in set_topic_partitions(); reorganize offsets…
Sep 15, 2014
67aa1a1
_should_auto_commit is private
Sep 15, 2014
8dc3623
raise KafkaConfigurationError in commit() if there is no configured '…
Sep 15, 2014
9dd7d7e
Add docstring to configure()
Sep 15, 2014
26e18ce
Add docstring to get_partition_offsets; use request_time_ms and max_n…
Sep 15, 2014
54ecfea
Add a few basic KafkaConsumer tests
Sep 16, 2014
5b88298
Use 4-space indents
Sep 16, 2014
08f6ad9
Reorder methods, add docstrings to public methds, section comments fo…
Sep 16, 2014
7caf9be
Add private methods to manage internal _msg_iter
Sep 16, 2014
07ff623
Move auto-commit checks to task_done; add support for auto_commit_int…
Sep 16, 2014
742af4f
Raise KafkaConfigurationError during fetch_messages if not topics/par…
Sep 16, 2014
391b532
Update docstrings w/ current interface / config defaults
Sep 16, 2014
6e6ae53
Use six for py3 support in KafkaConsumer
Sep 16, 2014
6b44828
Move KafkaConsumer to kafka.consumer.kafka module; make available for…
Sep 17, 2014
206560a
Force absolue_imports in kafka/consumer/kafka.py
Sep 17, 2014
b80e83b
Fix task_done checks when no previous commit exists; add test
Sep 17, 2014
9e1b203
Merge conflict w/ assertEqual (assertEquals deprecated)
Sep 17, 2014
1a06f79
Make TIMEOUT_MS configurable in test_kafka_consumer__blocking
Sep 17, 2014
08df93b
Add private methods _does_auto_commit_ms and _does_auto_commit_messages
Sep 22, 2014
209a8f2
OffsetCommit metadata must be bytes
Sep 22, 2014
ed893c3
Use kafka.util.kafka_bytestring to encode utf-8 when necessary
Dec 12, 2014
d27d49f
Fixup call to self._client.get_partition_ids_for_topic -- use encoded…
Dec 12, 2014
e3fd29c
Simplify BYTES_CONFIGURATION_KEYS logic, per wizzat review
Dec 12, 2014
29cae3e
Add some jitter to refresh_leader_backoff_ms, per wizzat review
Dec 12, 2014
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion kafka/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
)
from kafka.producer import SimpleProducer, KeyedProducer
from kafka.partitioner import RoundRobinPartitioner, HashedPartitioner
from kafka.consumer import SimpleConsumer, MultiProcessConsumer
from kafka.consumer import SimpleConsumer, MultiProcessConsumer, KafkaConsumer

__all__ = [
'KafkaClient', 'KafkaConnection', 'SimpleProducer', 'KeyedProducer',
Expand Down
6 changes: 3 additions & 3 deletions kafka/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,9 +119,9 @@ def _send_broker_unaware_request(self, payloads, encoder_fn, decoder_fn):
response = conn.recv(requestId)
return decoder_fn(response)

except Exception as e:
log.warning("Could not send request [%r] to server %s:%i, "
"trying next server: %s" % (requestId, host, port, e))
except Exception:
log.exception("Could not send request [%r] to server %s:%i, "
"trying next server" % (requestId, host, port))

raise KafkaUnavailableError("All servers failed to process request")

Expand Down
11 changes: 11 additions & 0 deletions kafka/common.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,9 @@
TopicAndPartition = namedtuple("TopicAndPartition",
["topic", "partition"])

KafkaMessage = namedtuple("KafkaMessage",
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So much yes to this. SO MUCH YES.

["topic", "partition", "offset", "key", "value"])


#################
# Exceptions #
Expand Down Expand Up @@ -182,6 +185,10 @@ class ConsumerNoMoreData(KafkaError):
pass


class ConsumerTimeout(KafkaError):
pass


class ProtocolError(KafkaError):
pass

Expand All @@ -190,6 +197,10 @@ class UnsupportedCodecError(KafkaError):
pass


class KafkaConfigurationError(KafkaError):
pass


kafka_errors = {
-1 : UnknownError,
0 : NoError,
Expand Down
3 changes: 2 additions & 1 deletion kafka/consumer/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
from .simple import SimpleConsumer
from .multiprocess import MultiProcessConsumer
from .kafka import KafkaConsumer

__all__ = [
'SimpleConsumer', 'MultiProcessConsumer'
'SimpleConsumer', 'MultiProcessConsumer', 'KafkaConsumer'
]
Loading