Skip to content

Allow injecting clients into Kafka Consumers and Producers. #735

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

Closed
wants to merge 3 commits into from
Closed

Allow injecting clients into Kafka Consumers and Producers. #735

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5868d31
Allow injecting clients into Kafka consumers and producers.
Julian Jun 21, 2016
c96ca37
Merge remote-tracking branch 'upstream/master'
Oct 4, 2018
b3db027
Merge pull request #1 from Magnetic/merge-latest-official-repo2
AndrewLvov Nov 6, 2018
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
12 changes: 10 additions & 2 deletions kafka/consumer/group.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,7 +46,10 @@ class KafkaConsumer(six.Iterator):
It just needs to have at least one broker that will respond to a
Metadata API Request. Default port is 9092. If no servers are
specified, will default to localhost:9092.
client_id (str): A name for this client. This string is passed in
client (kafka.client_async.KafkaClient): a kafka client to
use, or if unprovided, one is constructed from the provided
configuration.
client_id (str): a name for this client. This string is passed in
each request to servers and can be used to identify specific
server-side log entries that correspond to this client. Also
submitted to GroupCoordinator for logging with respect to
Expand Down Expand Up @@ -249,6 +252,7 @@ class KafkaConsumer(six.Iterator):
"""
DEFAULT_CONFIG = {
'bootstrap_servers': 'localhost',
'client': None,
'client_id': 'kafka-python-' + __version__,
'group_id': None,
'key_deserializer': None,
Expand Down Expand Up @@ -345,7 +349,11 @@ def __init__(self, *topics, **configs):
log.warning('use api_version=%s [tuple] -- "%s" as str is deprecated',
str(self.config['api_version']), str_version)

self._client = KafkaClient(metrics=self._metrics, **self.config)
client = self.config.pop('client', None) or KafkaClient(
metrics=self._metrics,
**self.config
)
self._client = client
Copy link
Owner

Choose a reason for hiding this comment

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

I'd probably just combine these into a one-line assignment


# Get auto-discovered version from client if necessary
if self.config['api_version'] is None:
Expand Down
11 changes: 9 additions & 2 deletions kafka/producer/kafka.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,9 @@ class KafkaProducer(object):
It just needs to have at least one broker that will respond to a
Metadata API Request. Default port is 9092. If no servers are
specified, will default to localhost:9092.
client (kafka.client_async.KafkaClient): a kafka client to
use, or if unprovided, one is constructed from the provided
configuration.
client_id (str): a name for this client. This string is passed in
each request to servers and can be used to identify specific
server-side log entries that correspond to this client.
Expand Down Expand Up @@ -279,6 +282,7 @@ class KafkaProducer(object):
"""
DEFAULT_CONFIG = {
'bootstrap_servers': 'localhost',
'client': None,
'client_id': None,
'key_serializer': None,
'value_serializer': None,
Expand Down Expand Up @@ -367,8 +371,11 @@ def __init__(self, **configs):
reporters = [reporter() for reporter in self.config['metric_reporters']]
self._metrics = Metrics(metric_config, reporters)

client = KafkaClient(metrics=self._metrics, metric_group_prefix='producer',
**self.config)
client = self.config['client'] or KafkaClient(
metrics=self._metrics,
metric_group_prefix='producer',
**self.config
)

# Get auto-discovered version from client if necessary
if self.config['api_version'] is None:
Expand Down