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 get_first_span_by_filter, testenv
 
 
@@ -29,7 +32,7 @@ def _resource(self) -> Generator[None, None, None]:
         self.kafka_client = AdminClient(self.kafka_config)
 
         try:
-            topics = self.kafka_client.create_topics(  # noqa: F841
+            _ = self.kafka_client.create_topics(  # noqa: F841
                 [
                     NewTopic(
                         testenv["kafka_topic"],
@@ -187,6 +190,130 @@ def test_trace_confluent_kafka_error(self) -> None:
             == "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) == 5
+
+        filtered_spans = agent.filter_spans(spans)
+        assert len(filtered_spans) == 3
+
+    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) == 5
+
+        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
+
     def test_confluent_kafka_consumer_root_exit(self) -> None:
         agent.options.allow_exit_as_root = True