fixes

Author: idanasulin2706

superclient/agent/interceptor.py

@@ -69,7 +69,9 @@ 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, "kafka-python")
+                opt_cfg, warning_msg = optimal_cfg(metadata, topics_env, orig_cfg, "kafka-python")
+                if warning_msg:
+                    error_msg = warning_msg
 
             # Apply optimized configuration
             for k, v in opt_cfg.items():
@@ -99,9 +101,6 @@ def init_patch(self, *args, **kwargs):
             )
             Heartbeat.register_tracker(tr)
 
-            # Initialize the producer with original configuration
-            orig_init(self, *args, **kwargs)
-
             # Patch send and close methods if not already patched
             if not hasattr(self, "_superstream_patch"):
                 original_send = self.send
@@ -125,7 +124,7 @@ def close_patch(*a, **kw):
                 self.close = close_patch
                 self._superstream_patch = True
 
-            # Initialize again with optimized configuration
+            # Initialize with optimized configuration
             orig_init(self, *args, **kwargs)
 
             # Send client registration message
@@ -191,7 +190,9 @@ 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, "aiokafka")
+                opt_cfg, warning_msg = optimal_cfg(metadata, topics_env, orig_cfg, "aiokafka")
+                if warning_msg:
+                    error_msg = warning_msg
             for k, v in opt_cfg.items():
                 current_val = kwargs.get(k)
                 if current_val != v:
@@ -214,7 +215,6 @@ def init_patch(self, *args, **kwargs):
                 topics_env=topics_env,
             )
             Heartbeat.register_tracker(tr)
-            orig_init(self, *args, **kwargs)
             if not hasattr(self, "_superstream_patch"):
                 original_send = self.send
 
@@ -294,7 +294,9 @@ 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, "confluent")
+                opt_cfg, warning_msg = optimal_cfg(metadata, topics_env, conf, "confluent")
+                if warning_msg:
+                    error_msg = warning_msg
             for k, v in opt_cfg.items():
                 current_val = conf.get(k)
                 if current_val != v:

superclient/agent/metadata.py

@@ -126,25 +126,40 @@ 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], lib_name: str) -> Dict[str, Any]:
-    """Compute optimal configuration based on metadata and topics."""
+def optimal_cfg(metadata: Optional[Dict[str, Any]], topics: list[str], orig: Dict[str, Any], lib_name: str) -> tuple[Dict[str, Any], str]:
+    """Compute optimal configuration based on metadata and topics.
+    
+    Returns:
+        tuple: (configuration_dict, warning_message)
+    """
     latency = os.getenv("SUPERSTREAM_LATENCY_SENSITIVE", "false").lower() == "true"
     cfg: Dict[str, Any]
+    warning_msg = ""
+    
     if not metadata or not metadata.get("topics_configuration"):
         logger.debug("No metadata or topics_configuration found; applying default configuration: %s", _DEFAULTS)
-        logger.warning("The topics you're publishing to haven't been analyzed yet. For optimal results, either wait for the next analysis cycle or trigger one manually via the SuperClient Console")
+        if not topics:
+            warning_msg = "No SUPERSTREAM_TOPICS_LIST environment variable set. Please set it to enable topic-specific optimizations."
+        else:
+            warning_msg = "The topics you're publishing to haven't been analyzed yet. For optimal results, either wait for the next analysis cycle or trigger one manually via the SuperClient Console"
+        logger.warning(warning_msg)
         cfg = dict(_DEFAULTS)
     else:
         matches = [tc for tc in metadata["topics_configuration"] if tc["topic_name"] in topics]
         if not matches:
             logger.debug("No matching topics found in metadata; applying default configuration: %s", _DEFAULTS)
-            logger.warning("The topics you're publishing to haven't been analyzed yet. For optimal results, either wait for the next analysis cycle or trigger one manually via the SuperClient Console")
+            if not topics:
+                warning_msg = "No SUPERSTREAM_TOPICS_LIST environment variable set. Please set it to enable topic-specific optimizations."
+            else:
+                warning_msg = "The topics you're publishing to haven't been analyzed yet. For optimal results, either wait for the next analysis cycle or trigger one manually via the SuperClient Console"
+            logger.warning(warning_msg)
             cfg = dict(_DEFAULTS)
         else:
             best = max(matches, key=lambda tc: tc["potential_reduction_percentage"] * tc["daily_writes_bytes"])
             cfg = dict(best.get("optimized_configuration", {}))
             for k, v in _DEFAULTS.items():
                 cfg.setdefault(k, v)
+    
     if latency:
         cfg.pop("linger.ms", None)
 
@@ -165,4 +180,4 @@ def optimal_cfg(metadata: Optional[Dict[str, Any]], topics: list[str], orig: Dic
                 pass
 
     # Translate Java-style keys to library-specific keys
-    return translate_java_to_lib(cfg, lib_name) 
+    return translate_java_to_lib(cfg, lib_name), warning_msg 

superclient/agent/tracker.py

@@ -52,7 +52,6 @@ def determine_topic(self) -> str:
         """Get the most impactful topic for this producer based on metadata analysis."""
         if not self.metadata or not self.metadata.get("topics_configuration"):
             # Fallback to first topic if no metadata available
-            logger.warning("No metadata available for producer {}, falling back to first used topic", self.client_id)
             return sorted(self.topics)[0] if self.topics else ""
 
         # Find matching topic configurations from metadata based on environment topics only
@@ -63,8 +62,6 @@ def determine_topic(self) -> str:
 
         if not matches:
             # Fallback to first environment topic if no matches
-            logger.warning("No matching topics found in metadata for producer {} (env topics: {}), falling back to first environment topic", 
-                          self.client_id, self.topics_env)
             return sorted(self.topics_env)[0] if self.topics_env else ""
 
         # Use the same logic as optimal_cfg: find the topic with highest impact

superclient/core/reporter.py

@@ -78,18 +78,26 @@ def internal_send_clients(bootstrap: str, base_cfg: Dict[str, Any], payload: byt
         return
 
     try:
+        # Handle aiokafka (async library)
         if lib_name == "aiokafka":
             asyncio.run(internal_send_clients_async(bootstrap, base_cfg, payload))
             return
 
-        prod = builder(bootstrap, base_cfg)
-        if hasattr(prod, "produce"):
-            prod.produce("superstream.clients", payload)
-            prod.flush()
-        else:
+        # Handle kafka-python (sync library)
+        if lib_name == "kafka-python":
+            prod = builder(bootstrap, base_cfg)
             prod.send("superstream.clients", payload)
             prod.flush()
             prod.close()
+            return
+
+        # Handle confluent-kafka (sync library with different API)
+        if lib_name == "confluent":
+            prod = builder(bootstrap, base_cfg)
+            prod.produce("superstream.clients", payload)
+            prod.flush()
+            return
+
     except Exception:
         logger.debug("Failed to send clients message via {}", lib_name)
 

superclient/util/config.py

@@ -30,7 +30,27 @@ def get_topics_list() -> List[str]:
 
 def mask_sensitive(k: str, v: Any) -> Any:
     """Mask sensitive configuration values."""
-    return "[MASKED]" if "password" in k.lower() or "sasl.jaas.config" in k.lower() else v
+    sensitive_patterns = [
+        "password", "sasl.jaas.config", "basic.auth.user.info",
+        "ssl.key", "ssl.cert", "ssl.truststore", "ssl.keystore",
+        "sasl.kerberos.keytab", "sasl.kerberos.principal"
+    ]
+    return "[MASKED]" if any(pattern in k.lower() for pattern in sensitive_patterns) else v
+
+def _serialize_config_value(v: Any) -> Any:
+    """Convert configuration values to JSON-serializable format."""
+    if callable(v):
+        # Convert functions to string representation
+        return f"<function: {v.__name__ if hasattr(v, '__name__') else str(v)}>"
+    elif hasattr(v, '__dict__'):
+        # Handle objects that might have __dict__ but aren't functions
+        try:
+            # Try to serialize as dict, fallback to string representation
+            return v.__dict__
+        except:
+            return f"<object: {type(v).__name__}>"
+    else:
+        return v
 
 def copy_client_configuration_properties(src: Dict[str, Any], dst: Dict[str, Any]):
     """Copy essential client configuration properties from source to destination.
@@ -137,7 +157,6 @@ def copy_client_configuration_properties(src: Dict[str, Any], dst: Dict[str, Any
         "sasl.kerberos.domain.name": "sasl_kerberos_domain_name",
         "sasl.oauth.token.provider": "sasl_oauth_token_provider",
         "socks5.proxy": "socks5_proxy",
-        "compression.codec": "compression_type",  # Maps to compression_type
     },
     "aiokafka": {
         # Basic configuration
@@ -159,7 +178,6 @@ def copy_client_configuration_properties(src: Dict[str, Any], dst: Dict[str, Any
         "enable.idempotence": "enable_idempotence",
         "security.protocol": "security_protocol",
         "sasl.mechanism": "sasl_mechanism",
-        "compression.codec": "compression_type",  # Maps to compression_type
     },
     "confluent": {
         # Confluent uses Java-style names directly, so most mappings are 1:1
@@ -256,7 +274,7 @@ def translate_lib_to_java(cfg: Dict[str, Any], lib_name: str) -> Dict[str, Any]:
         "enable_idempotence": False,
         "delivery_timeout_ms": 120000,
         "acks": 1,
-        "compression_type": None,
+        "compression_type": "none",
         "retries": 0,
         "batch_size": 16384,
         "linger_ms": 0,
@@ -303,7 +321,7 @@ def translate_lib_to_java(cfg: Dict[str, Any], lib_name: str) -> Dict[str, Any]:
 
         # Producer specific
         "acks": 1,
-        "compression_type": None,
+        "compression_type": "none",
         "max_batch_size": 16384,  # aiokafka uses max_batch_size
         "linger_ms": 0,
         "partitioner": None,
@@ -422,5 +440,10 @@ def get_original_config(orig_cfg: Dict[str, Any], lib_name: str) -> Dict[str, An
     for k, v in defaults_java.items():
         if k not in user_keys_java:
             merged[k] = v
+    
+    # Serialize any function objects to make them JSON-serializable
+    serialized: Dict[str, Any] = {}
+    for k, v in merged.items():
+        serialized[k] = _serialize_config_value(v)
 
-    return merged
+    return serialized