reset client offsets

kafka_consumer/datadog_checks/kafka_consumer/client/kafka_python_client.py

@@ -415,3 +415,7 @@ def get_partitions_for_topic(self, topic):
 
     def request_metadata_update(self):
         self.kafka_client._client.cluster.request_update()
+
+    def reset_offsets(self):
+        self._consumer_offsets = {}
+        self._highwater_offsets = {}

kafka_consumer/datadog_checks/kafka_consumer/kafka_consumer.py

@@ -40,6 +40,8 @@ def __init__(self, name, init_config, instances):
     def check(self, _):
         """The main entrypoint of the check."""
         # Fetch Kafka consumer offsets
+        self.client.reset_offsets()
+
         try:
             self.client.get_consumer_offsets()
         except Exception:
@@ -158,7 +160,7 @@ def collect_broker_metadata(self):
         version_data = [str(part) for part in self.client.collect_broker_version()]
         version_parts = {name: part for name, part in zip(('major', 'minor', 'patch'), version_data)}
 
-        self.check.set_metadata(
+        self.set_metadata(
             'version', '.'.join(version_data), scheme='parts', final_scheme='semver', part_map=version_parts
         )