Handle empty topics and partitions

kafka_consumer/datadog_checks/kafka_consumer/data/conf.yaml.example

@@ -78,18 +78,16 @@ instances:
     ## @param consumer_groups - object - optional
     ## When using Kafka to store consumer offsets each level is mandatory.
     ## When using zookeeper to store consumer offsets  each level is optional.
-    ## Any omitted values are fetched from Zookeeper. You can omit partitions (example: <CONSUMER_NAME_2>),
+    ## Any empty values are fetched from Zookeeper. You can have empty partitions (example: <CONSUMER_NAME_2>),
     ## topics (example: <CONSUMER_NAME_3>), and even consumer_groups. If you omit
     ## consumer_groups, you must set 'monitor_unlisted_consumer_groups': True.
-    ## If a value is omitted, the parent value must still be it's expected type,
-    ## which is typically a dict.
     #
     # consumer_groups:
     #   <CONSUMER_NAME_1>:
     #     <TOPIC_NAME_1>: [0, 1, 4, 12]
     #   <CONSUMER_NAME_2>:
-    #     <TOPIC_NAME_2>:
-    #   <CONSUMER_NAME_3>
+    #     <TOPIC_NAME_2>: []
+    #   <CONSUMER_NAME_3>: {}
 
     ## @param monitor_unlisted_consumer_groups - boolean - required
     ## Setting monitor_unlisted_consumer_groups to True tells the check to

kafka_consumer/datadog_checks/kafka_consumer/kafka_consumer.py

@@ -374,7 +374,8 @@ def _report_consumer_metrics(self, highwater_offsets, consumer_offsets, unled_to
                 'topic:%s' % topic,
                 'partition:%s' % partition,
                 'consumer_group:%s' % consumer_group,
-            ] + tags
+            ]
+            consumer_group_tags.extend(tags)
             self.gauge('kafka.consumer_offset', consumer_offset, tags=consumer_group_tags)
 
             consumer_lag = highwater_offsets[(topic, partition)] - consumer_offset
@@ -435,14 +436,14 @@ def _get_zk_consumer_offsets(self, zk_hosts_ports, consumer_groups=None, zk_pref
                 }
 
             for consumer_group, topics in iteritems(consumer_groups):
-                if topics is None:
+                if not topics:
                     # If topics are't specified, fetch them from ZK
                     zk_path_topics = zk_path_topic_tmpl.format(group=consumer_group)
                     topics = {topic: None for topic in self._get_zk_path_children(zk_conn, zk_path_topics, 'topics')}
                     consumer_groups[consumer_group] = topics
 
                 for topic, partitions in iteritems(topics):
-                    if partitions is not None:
+                    if partitions:
                         partitions = set(partitions)  # defend against bad user input
                     else:
                         # If partitions aren't specified, fetch them from ZK

kafka_consumer/tests/test_kafka_consumer_zk.py

@@ -10,7 +10,7 @@
 from datadog_checks.base.utils.containers import hash_mutable
 from datadog_checks.kafka_consumer import KafkaCheck
 
-from .common import HOST, PARTITIONS, TOPICS, ZK_CONNECT_STR, is_supported
+from .common import KAFKA_CONNECT_STR, HOST, PARTITIONS, TOPICS, ZK_CONNECT_STR, is_supported
 
 pytestmark = pytest.mark.skipif(
     not is_supported('zookeeper'), reason='zookeeper consumer offsets not supported in current environment'
@@ -45,6 +45,19 @@ def test_check_zk(aggregator, zk_instance):
     aggregator.assert_metric('kafka.broker_offset', at_least=1)
     aggregator.assert_all_metrics_covered()
 
+    all_partitions = {
+        'kafka_connect_str': KAFKA_CONNECT_STR,
+        'zk_connect_str': ZK_CONNECT_STR,
+        'consumer_groups': {'my_consumer': {'marvel': []}},
+    }
+    kafka_consumer_check.check(all_partitions)
+    aggregator.assert_metric(
+        'kafka.consumer_offset', tags=['topic:marvel', 'partition:0',
+            'consumer_group:my_consumer', 'source:zk'], at_least=1)
+    aggregator.assert_metric(
+        'kafka.consumer_offset', tags=['topic:marvel', 'partition:1',
+            'consumer_group:my_consumer', 'source:zk'], at_least=1)
+
 
 @pytest.mark.usefixtures('dd_environment', 'kafka_producer', 'zk_consumer')
 def test_multiple_servers_zk(aggregator, zk_instance):

kafka_consumer/tox.ini

@@ -16,7 +16,7 @@ deps =
     -rrequirements-dev.txt
 commands =
     pip install -r requirements.in
-    pytest -v
+    pytest -v {posargs}
 setenv =
     kafka: KAFKA_OFFSETS_STORAGE=kafka
     zk: KAFKA_OFFSETS_STORAGE=zookeeper