general improvements

Author: idanasulin2706

superclient/agent/clients.py

@@ -69,7 +69,7 @@ def init_patch(self, *args, **kwargs):
                 opt_cfg = {}
             else:
                 # Get optimized configuration if Superstream is active
-                opt_cfg = optimal_cfg(metadata, topics_env, orig_cfg)
+                opt_cfg = optimal_cfg(metadata, topics_env, orig_cfg, "kafka-python")
 
             # Apply optimized configuration
             for k, v in opt_cfg.items():
@@ -91,6 +91,8 @@ def init_patch(self, *args, **kwargs):
                 opt_cfg=opt_cfg,
                 report_interval_ms=int(report_interval or _DEFAULT_REPORT_INTERVAL_MS),
                 error=error_msg,  # Store error message in tracker
+                metadata=metadata,
+                topics_env=topics_env,
             )
             Heartbeat.register_tracker(tr)
 
@@ -186,7 +188,7 @@ def init_patch(self, *args, **kwargs):
                 opt_cfg = {}
             else:
                 # Get optimized configuration if Superstream is active
-                opt_cfg = optimal_cfg(metadata, topics_env, orig_cfg)
+                opt_cfg = optimal_cfg(metadata, topics_env, orig_cfg, "aiokafka")
             for k, v in opt_cfg.items():
                 if kwargs.get(k) != v:
                     logger.debug("Overriding configuration: {} -> {}", k, v)
@@ -201,6 +203,8 @@ def init_patch(self, *args, **kwargs):
                 opt_cfg=opt_cfg,
                 report_interval_ms=int(report_interval or _DEFAULT_REPORT_INTERVAL_MS),
                 error=error_msg,  # Store error message in tracker
+                metadata=metadata,
+                topics_env=topics_env,
             )
             Heartbeat.register_tracker(tr)
             orig_init(self, *args, **kwargs)
@@ -283,7 +287,7 @@ def init_patch(self, conf: Dict[str, Any], *args, **kwargs):
                 opt_cfg = {}
             else:
                 # Get optimized configuration if Superstream is active
-                opt_cfg = optimal_cfg(metadata, topics_env, conf)
+                opt_cfg = optimal_cfg(metadata, topics_env, conf, "confluent")
             for k, v in opt_cfg.items():
                 if conf.get(k) != v:
                     logger.debug("Overriding configuration: {} -> {}", k, v)
@@ -298,6 +302,8 @@ def init_patch(self, conf: Dict[str, Any], *args, **kwargs):
                 opt_cfg=opt_cfg,
                 report_interval_ms=int(report_interval or _DEFAULT_REPORT_INTERVAL_MS),
                 error=error_msg,  # Store error message in tracker
+                metadata=metadata,
+                topics_env=topics_env,
             )
             Heartbeat.register_tracker(tr)
             orig_init(self, conf, *args, **kwargs)

superclient/agent/interceptor.py

@@ -5,7 +5,7 @@
 from typing import Any, Dict, Optional, Literal
 
 from ..util.logger import get_logger
-from ..util.config import copy_client_configuration_properties
+from ..util.config import copy_client_configuration_properties, translate_java_to_lib
 
 logger = get_logger("agent.metadata")
 
@@ -126,7 +126,7 @@ def fetch_metadata(
         logger.error("[ERR-203] Failed to fetch metadata: {}", exc)
     return None
 
-def optimal_cfg(metadata: Optional[Dict[str, Any]], topics: list[str], orig: Dict[str, Any]) -> Dict[str, Any]:
+def optimal_cfg(metadata: Optional[Dict[str, Any]], topics: list[str], orig: Dict[str, Any], lib_name: str) -> Dict[str, Any]:
     """Compute optimal configuration based on metadata and topics."""
     latency = os.getenv("SUPERSTREAM_LATENCY_SENSITIVE", "false").lower() == "true"
     cfg: Dict[str, Any]
@@ -154,4 +154,6 @@ def optimal_cfg(metadata: Optional[Dict[str, Any]], topics: list[str], orig: Dic
                     cfg[p] = orig[p]
             except Exception:
                 pass
-    return cfg 
+    
+    # Translate Java-style keys to library-specific keys
+    return translate_java_to_lib(cfg, lib_name)