Skip to content
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

Jump to bottom

Implement get_partitions_for_topic #14079

Merged
merged 3 commits into from
Mar 6, 2023
Merged

Implement get_partitions_for_topic #14079

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
90ccbda
Implement get_partitions_for_topic
yzhan289 Mar 1, 2023
579fe1e
Add exception handling
yzhan289 Mar 3, 2023
119c6ec
Fix style
yzhan289 Mar 6, 2023
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
11 changes: 10 additions & 1 deletion kafka_consumer/datadog_checks/kafka_consumer/client/confluent_kafka_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
# (C) Datadog, Inc. 2023-present
# All rights reserved
# Licensed under a 3-clause BSD style license (see LICENSE)
from confluent_kafka import KafkaException
from confluent_kafka.admin import AdminClient

from datadog_checks.kafka_consumer.client.kafka_client import KafkaClient
Expand Down Expand Up @@ -35,7 +36,15 @@ def reset_offsets(self):
raise NotImplementedError

def get_partitions_for_topic(self, topic):
raise NotImplementedError

try:
cluster_metadata = self.kafka_client.list_topics(topic)
topic_metadata = cluster_metadata.topics[topic]
partitions = list(topic_metadata.partitions.keys())
return partitions
except KafkaException as e:
self.log.error("Received exception when getting partitions for topic %s: %s", topic, e)
return None
Copy link
Contributor Author

Choose a reason for hiding this comment

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

To reviewer: I am returning None here since the kafka-python implementation of get_partitions_for_topic() returns this in the worst case.


def request_metadata_update(self):
raise NotImplementedError
Expand Down
8 changes: 3 additions & 5 deletions kafka_consumer/datadog_checks/kafka_consumer/client/generic_kafka_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,11 +46,9 @@ def reset_offsets(self):
return self.python_kafka_client.reset_offsets()

def get_partitions_for_topic(self, topic):
# TODO when this method is implemented in ConfluentKafkaClient, replace this with:
# if self.use_legacy_client:
# return self.python_kafka_client.get_partitions_for_topic(topic)
# return self.confluent_kafka_client.get_partitions_for_topic(topic)
return self.python_kafka_client.get_partitions_for_topic(topic)
if self.use_legacy_client:
return self.python_kafka_client.get_partitions_for_topic(topic)
return self.confluent_kafka_client.get_partitions_for_topic(topic)

def request_metadata_update(self):
# TODO when this method is implemented in ConfluentKafkaClient, replace this with:
Expand Down