feat: added span filtering for confluent-kafka module

Signed-off-by: Cagri Yonca <[email protected]>

Author: CagriYonca

tests/clients/kafka/test_confluent_kafka.py

@@ -1,5 +1,6 @@
 # (c) Copyright IBM Corp. 2025
 
+import os
 from typing import Generator
 
 import pytest
@@ -11,7 +12,9 @@
 from confluent_kafka.admin import AdminClient, NewTopic
 from opentelemetry.trace import SpanKind
 
+from instana.options import StandardOptions
 from instana.singletons import agent, tracer
+from instana.util.config import parse_ignored_endpoints_from_yaml
 from tests.helpers import testenv
 
 
@@ -186,3 +189,127 @@ def test_trace_confluent_kafka_error(self) -> None:
             kafka_span.data["kafka"]["error"]
             == "num_messages must be between 0 and 1000000 (1M)"
         )
+
+    def test_ignore_confluent_kafka(self) -> None:
+        os.environ["INSTANA_IGNORE_ENDPOINTS"] = "kafka"
+        agent.options = StandardOptions()
+
+        with tracer.start_as_current_span("test"):
+            self.producer.produce(testenv["kafka_topic"], b"raw_bytes")
+            self.producer.flush(timeout=10)
+
+        spans = self.recorder.queued_spans()
+        assert len(spans) == 2
+
+        filtered_spans = agent.filter_spans(spans)
+        assert len(filtered_spans) == 1
+
+    def test_ignore_confluent_kafka_producer(self) -> None:
+        os.environ["INSTANA_IGNORE_ENDPOINTS"] = "kafka:produce"
+        agent.options = StandardOptions()
+
+        with tracer.start_as_current_span("test-span"):
+            # Produce some events
+            self.producer.produce(testenv["kafka_topic"], b"raw_bytes1")
+            self.producer.produce(testenv["kafka_topic"], b"raw_bytes2")
+            self.producer.flush()
+
+            # Consume the events
+            consumer_config = self.kafka_config.copy()
+            consumer_config["group.id"] = "my-group"
+            consumer_config["auto.offset.reset"] = "earliest"
+
+            consumer = Consumer(consumer_config)
+            consumer.subscribe([testenv["kafka_topic"]])
+            consumer.consume(num_messages=2, timeout=60)
+
+        consumer.close()
+
+        spans = self.recorder.queued_spans()
+        assert len(spans) == 4
+
+        filtered_spans = agent.filter_spans(spans)
+        assert len(filtered_spans) == 2
+
+    def test_ignore_confluent_kafka_consumer(self) -> None:
+        os.environ["INSTANA_IGNORE_ENDPOINTS"] = "kafka:consume"
+        agent.options = StandardOptions()
+
+        with tracer.start_as_current_span("test-span"):
+            # Produce some events
+            self.producer.produce(testenv["kafka_topic"], b"raw_bytes1")
+            self.producer.produce(testenv["kafka_topic"], b"raw_bytes2")
+            self.producer.flush()
+
+            # Consume the events
+            consumer_config = self.kafka_config.copy()
+            consumer_config["group.id"] = "my-group"
+            consumer_config["auto.offset.reset"] = "earliest"
+
+            consumer = Consumer(consumer_config)
+            consumer.subscribe([testenv["kafka_topic"]])
+            consumer.consume(num_messages=2, timeout=60)
+
+        consumer.close()
+
+        spans = self.recorder.queued_spans()
+        assert len(spans)
+
+        filtered_spans = agent.filter_spans(spans)
+        assert len(filtered_spans) == 3
+
+    def test_ignore_confluent_specific_topic(self) -> None:
+        os.environ["INSTANA_IGNORE_ENDPOINTS"] = "kafka:consume"
+        os.environ["INSTANA_IGNORE_ENDPOINTS_PATH"] = (
+            "tests/util/test_configuration-1.yaml"
+        )
+
+        agent.options = StandardOptions()
+
+        with tracer.start_as_current_span("test-span"):
+            # Produce some events
+            self.producer.produce(testenv["kafka_topic"], b"raw_bytes1")
+            self.producer.flush()
+
+            # Consume the events
+            consumer_config = self.kafka_config.copy()
+            consumer_config["group.id"] = "my-group"
+            consumer_config["auto.offset.reset"] = "earliest"
+
+            consumer = Consumer(consumer_config)
+            consumer.subscribe([testenv["kafka_topic"]])
+            consumer.consume(num_messages=1, timeout=60)
+
+        consumer.close()
+
+        spans = self.recorder.queued_spans()
+        assert len(spans) == 3
+
+        filtered_spans = agent.filter_spans(spans)
+        assert len(filtered_spans) == 1
+
+    def test_ignore_confluent_specific_topic_with_config_file(self) -> None:
+        agent.options.ignore_endpoints = parse_ignored_endpoints_from_yaml(
+            "tests/util/test_configuration-1.yaml"
+        )
+
+        with tracer.start_as_current_span("test-span"):
+            # Produce some events
+            self.producer.produce(testenv["kafka_topic"], b"raw_bytes1")
+            self.producer.flush()
+
+            # Consume the events
+            consumer_config = self.kafka_config.copy()
+            consumer_config["group.id"] = "my-group"
+            consumer_config["auto.offset.reset"] = "earliest"
+
+            consumer = Consumer(consumer_config)
+            consumer.subscribe([testenv["kafka_topic"]])
+            consumer.consume(num_messages=1, timeout=60)
+        consumer.close()
+
+        spans = self.recorder.queued_spans()
+        assert len(spans) == 3
+
+        filtered_spans = agent.filter_spans(spans)
+        assert len(filtered_spans) == 1