Collect version metadata

kafka_consumer/datadog_checks/kafka_consumer/kafka_consumer.py

@@ -114,6 +114,8 @@ def check(self, instance):
         self._report_highwater_offsets()
         self._report_consumer_offsets_and_lag()
 
+        self._collect_broker_metadata()
+
     def _create_kafka_admin_client(self, api_version):
         """Return a KafkaAdminClient."""
         kafka_connect_str = self.instance.get('kafka_connect_str')
@@ -433,6 +435,15 @@ def _send_event(self, title, text, tags, event_type, aggregation_key, severity='
         }
         self.event(event_dict)
 
+    @AgentCheck.metadata_entrypoint
+    def _collect_broker_metadata(self):
+        version_data = [str(part) for part in self._kafka_client._client.check_version()]
+        version_parts = {name: part for name, part in zip(('major', 'minor', 'patch'), version_data)}
+
+        self.set_metadata(
+            'version', '.'.join(version_data), scheme='parts', final_scheme='semver', part_map=version_parts
+        )
+
     @classmethod
     def _determine_kafka_version(cls, init_config, instance):
         """Return the Kafka cluster version as a tuple."""