fixes

Author: idanasulin2706

.gitignore

@@ -4,4 +4,5 @@ __pycache__/
 *$py.class
 superclient.egg-info
 build/
-dist/
+dist/
+superstream_clients.egg-info/

superclient/agent/__init__.py

@@ -19,9 +19,6 @@
     set_debug_enabled(True)
 
 logger = get_logger("agent")
-logger.info("Superstream Agent initialized with environment variables: {}", _ENV_VARS)
-if is_disabled():
-    logger.warn("Superstream functionality disabled via SUPERSTREAM_DISABLED")
 
 # Preserve reference to built-in import function
 _original_import = builtins.__import__
@@ -93,6 +90,12 @@ def initialize():
     2. Schedules patching of any pre-imported modules
     3. Starts the heartbeat thread
     """
+    
+    # Log initialization message
+    logger.info("Superstream Agent initialized with environment variables: {}", _ENV_VARS)
+    if is_disabled():
+        logger.warn("Superstream functionality disabled via SUPERSTREAM_DISABLED")
+    
     if is_disabled():
         return
 

superclient/agent/interceptor.py

@@ -62,7 +62,7 @@ def init_patch(self, *args, **kwargs):
                 error_msg = "[ERR-304] Failed to fetch metadata for producer with client id {}: Unable to connect to Superstream service".format(client_id)
                 # Skip optimization but keep stats reporting
                 opt_cfg = {}
-            elif not metadata.active:
+            elif not metadata.get("active", True):
                 error_msg = "[ERR-301] Superstream optimization is not active for this kafka cluster, please head to the Superstream console and activate it."
                 logger.error(error_msg)
                 # Skip optimization but keep stats reporting
@@ -73,13 +73,12 @@ def init_patch(self, *args, **kwargs):
 
             # Apply optimized configuration
             for k, v in opt_cfg.items():
-                snake = k.replace(".", "_")
-                if kwargs.get(snake) != v:
-                    logger.debug("Overriding configuration: {} -> {}", snake, v)
-                    kwargs[snake] = v
+                if kwargs.get(k) != v:
+                    logger.debug("Overriding configuration: {} -> {}", k, v)
+                    kwargs[k] = v
 
             # Set up reporting interval
-            report_interval = metadata.report_interval_ms if metadata else _DEFAULT_REPORT_INTERVAL_MS
+            report_interval = metadata.get("report_interval_ms") if metadata else _DEFAULT_REPORT_INTERVAL_MS
 
             # Create and register producer tracker
             tr = ProducerTracker(
@@ -181,7 +180,7 @@ def init_patch(self, *args, **kwargs):
                 logger.error(error_msg)
                 # Skip optimization but keep stats reporting
                 opt_cfg = {}
-            elif not metadata.active:
+            elif not metadata.get("active", True):
                 error_msg = "[ERR-301] Superstream optimization is not active for this kafka cluster, please head to the Superstream console and activate it."
                 logger.error(error_msg)
                 # Skip optimization but keep stats reporting
@@ -193,7 +192,7 @@ def init_patch(self, *args, **kwargs):
                 if kwargs.get(k) != v:
                     logger.debug("Overriding configuration: {} -> {}", k, v)
                     kwargs[k] = v
-            report_interval = metadata.report_interval_ms if metadata else _DEFAULT_REPORT_INTERVAL_MS
+            report_interval = metadata.get("report_interval_ms") if metadata else _DEFAULT_REPORT_INTERVAL_MS
             tr = ProducerTracker(
                 lib="aiokafka",
                 producer=self,
@@ -280,7 +279,7 @@ def init_patch(self, conf: Dict[str, Any], *args, **kwargs):
                 logger.error(error_msg)
                 # Skip optimization but keep stats reporting
                 opt_cfg = {}
-            elif not metadata.active:
+            elif not metadata.get("active", True):
                 error_msg = "[ERR-301] Superstream optimization is not active for this kafka cluster, please head to the Superstream console and activate it."
                 logger.error(error_msg)
                 # Skip optimization but keep stats reporting
@@ -292,7 +291,7 @@ def init_patch(self, conf: Dict[str, Any], *args, **kwargs):
                 if conf.get(k) != v:
                     logger.debug("Overriding configuration: {} -> {}", k, v)
                     conf[k] = v
-            report_interval = metadata.report_interval_ms if metadata else _DEFAULT_REPORT_INTERVAL_MS
+            report_interval = metadata.get("report_interval_ms") if metadata else _DEFAULT_REPORT_INTERVAL_MS
             tr = ProducerTracker(
                 lib="confluent",
                 producer=self,