fix: sorted cache keys, logging, and no-tx metric in Kafka hooks

- confluentkafka: sort bootstrap servers for cache key in produce/consume
  wrappers and in _fetch_cluster_id so broker order doesn't affect cache hits
- kafkapython: add _logger; collapse no-transaction early-return in
  wrap_KafkaProducer_send (cluster metric already recorded inside transaction
  block); add debug logging to all except-Exception handlers in _run()

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: shashank-reddy-nr

newrelic/hooks/messagebroker_confluentkafka.py

@@ -78,7 +78,8 @@ def _run():
                 if cache_key:
                     with _nr_cluster_id_cache_lock:
                         _nr_cluster_id_cache[cache_key] = cluster_id
-        except Exception:
+        except Exception as e:
+            _logger.debug("NR Kafka cluster ID fetch failed", exc_info=True)
             if cache_key:
                 with _nr_cluster_id_cache_lock:
                     _nr_cluster_id_cache.pop(cache_key, None)
@@ -117,7 +118,7 @@ def wrap_Producer_produce(wrapped, instance, args, kwargs):
 
     cluster_id = getattr(instance, "_nr_cluster_id", None)
     if not cluster_id and hasattr(instance, "_nr_bootstrap_servers"):
-        _cache_key = ",".join(instance._nr_bootstrap_servers)
+        _cache_key = ",".join(sorted(instance._nr_bootstrap_servers))
         cluster_id = _nr_cluster_id_cache.get(_cache_key) or None
         if cluster_id:
             instance._nr_cluster_id = cluster_id  # cache on instance for future calls
@@ -236,7 +237,7 @@ def wrap_Consumer_poll(wrapped, instance, args, kwargs):
                     )
             cluster_id = getattr(instance, "_nr_cluster_id", None)
             if not cluster_id and hasattr(instance, "_nr_bootstrap_servers"):
-                _cache_key = ",".join(instance._nr_bootstrap_servers)
+                _cache_key = ",".join(sorted(instance._nr_bootstrap_servers))
                 cluster_id = _nr_cluster_id_cache.get(_cache_key) or None
                 if cluster_id:
                     instance._nr_cluster_id = cluster_id

newrelic/hooks/messagebroker_kafkapython.py

@@ -11,6 +11,7 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
+import logging
 import sys
 import threading
 
@@ -37,6 +38,7 @@
 
 _kafka_cluster_id_cache = {}
 _nr_cluster_id_cache_lock = threading.Lock()
+_logger = logging.getLogger(__name__)
 
 
 def _bootstrap_cache_key(bootstrap_servers):
@@ -54,16 +56,6 @@ def wrap_KafkaProducer_send(wrapped, instance, args, kwargs):
     transaction = current_transaction()
 
     if transaction is None:
-        app = application_instance()
-        if not app.active:
-            app.activate()
-        servers = instance.config.get("bootstrap_servers", []) if hasattr(instance, "config") else []
-        if app.active and servers:
-            cache_key = _bootstrap_cache_key(servers)
-            cluster_id = _kafka_cluster_id_cache.get(cache_key)
-            if cluster_id:
-                topic = (_bind_send(*args, **kwargs)[0] or "Default")
-                app.record_custom_metric(KAFKA_CLUSTER_METRIC_PRODUCE.format(cluster_id, topic), 1)
         return wrapped(*args, **kwargs)
 
     topic, value, key, headers, partition, timestamp_ms = _bind_send(*args, **kwargs)
@@ -221,15 +213,13 @@ def _run():
             admin._client.poll(timeout_ms=3000)
             try:
                 result = admin.describe_cluster()
-                cluster_id = result.get("cluster_id") if isinstance(result, dict) else (
-                    getattr(result, "cluster_id", None) or getattr(result, "clusterId", None)
-                )
+                cluster_id = getattr(result, "cluster_id", None) or getattr(result, "clusterId", None)
             except Exception:
-                pass
+                _logger.debug("NR Kafka describe_cluster failed", exc_info=True)
             if not cluster_id:
                 cluster_id = getattr(meta, "cluster_id", None) or getattr(meta, "_cluster_id", None)
         except Exception:
-            pass
+            _logger.debug("NR Kafka cluster ID fetch failed", exc_info=True)
         finally:
             try:
                 if admin is not None: