kafka_consumer: lowercase Kafka cluster in data streams messages feature (DataDog#20842)

Author: piochelepiotr

kafka_consumer/datadog_checks/kafka_consumer/kafka_consumer.py

@@ -418,7 +418,7 @@ def data_streams_live_message(self, highwater_offsets, cluster_id):
             config_id = cfg["id"]
             if self._messages_have_been_retrieved(config_id):
                 continue
-            if cluster != cluster_id:
+            if not cluster or not cluster_id or cluster.lower() != cluster_id.lower():
                 continue
             start_offsets = resolve_start_offsets(highwater_offsets, topic, partition, start_offset, n_messages)
 

kafka_consumer/tests/test_unit.py

@@ -26,15 +26,15 @@ def fake_consumer_offsets_for_times(partitions):
     return [(t, p, 80) for t, p in partitions]
 
 
-def seed_mock_client():
+def seed_mock_client(cluster_id="cluster_id"):
     """Set some common defaults for the mock client to kafka."""
     client = mock.create_autospec(KafkaClient)
     client.list_consumer_groups.return_value = ["consumer_group1"]
     client.get_partitions_for_topic.return_value = ['partition1']
     client.list_consumer_group_offsets.return_value = [("consumer_group1", [("topic1", "partition1", 2)])]
     client.describe_consumer_group.return_value = 'STABLE'
     client.consumer_get_cluster_id_and_list_topics.return_value = (
-        "cluster_id",
+        cluster_id,
         # topics
         [
             # Used in unit tets
@@ -618,7 +618,7 @@ def test_data_streams_messages(
             }
         ),
     )
-    mock_client = seed_mock_client()
+    mock_client = seed_mock_client(cluster_id="Cluster_id")
     mock_client.get_next_message.side_effect = [
         MockedMessage(
             b'{"name": "Peter Parker", "age": 18, "transaction_amount": 123, "currency": "dollar"}',