fixes

Author: idanasulin2706

examples/requirements.txt

@@ -1,4 +1,4 @@
-kafka-python==2.0.2
+kafka-python==2.2.14
 confluent-kafka==2.3.0
 aiokafka==0.10.0 
 aws-msk-iam-sasl-signer-python==1.0.2

superclient/agent/metadata.py

@@ -24,7 +24,7 @@ def _create_consumer_kafka_python(bootstrap: str, base_cfg: Dict[str, Any]):
         "enable_auto_commit": False,
         "auto_offset_reset": "earliest",
     }
-    copy_client_configuration_properties(base_cfg, consumer_cfg)
+    copy_client_configuration_properties(base_cfg, consumer_cfg, "kafka-python")
     return kafka.KafkaConsumer(**consumer_cfg)
 
 def _create_consumer_aiokafka(bootstrap: str, base_cfg: Dict[str, Any]):
@@ -36,7 +36,7 @@ def _create_consumer_aiokafka(bootstrap: str, base_cfg: Dict[str, Any]):
         "enable_auto_commit": False,
         "auto_offset_reset": "earliest",
     }
-    copy_client_configuration_properties(base_cfg, consumer_cfg)
+    copy_client_configuration_properties(base_cfg, consumer_cfg, "aiokafka")
     return AIOKafkaConsumer(**consumer_cfg)
 
 def _create_consumer_confluent(bootstrap: str, base_cfg: Dict[str, Any]):
@@ -48,7 +48,7 @@ def _create_consumer_confluent(bootstrap: str, base_cfg: Dict[str, Any]):
         "enable.auto.commit": False,
         "auto.offset.reset": "earliest",
     }
-    copy_client_configuration_properties(base_cfg, consumer_cfg)
+    copy_client_configuration_properties(base_cfg, consumer_cfg, "confluent")
     return Consumer(consumer_cfg)
 
 _CONSUMER_BUILDERS = {
@@ -108,14 +108,18 @@ def fetch_metadata(
 
             tp = _kafka.TopicPartition(topic, 0)
             consumer.assign([tp])
-            consumer.seek_to_end(tp)
-            end = consumer.position(tp)
+            
+            # Get the end offset safely
+            end_offsets = consumer.end_offsets([tp])
+            end = end_offsets.get(tp, 0)
+            
             if end == 0:
                 logger.error(
                     "[ERR-202] Unable to retrieve optimizations data from Superstream – topic empty."
                 )
                 consumer.close()
                 return None
+                
             consumer.seek(tp, end - 1)
             recs = consumer.poll(timeout_ms=5000)
             consumer.close()

superclient/core/manager.py

@@ -60,7 +60,7 @@ def fetch_metadata_kafka_python(bootstrap: str, cfg: Dict[str, Any]) -> Optional
             "enable_auto_commit": False,
             "group_id": "superstream-metadata-fetcher",
         }
-        copy_client_configuration_properties(cfg, kafka_cfg)
+        copy_client_configuration_properties(cfg, kafka_cfg, "kafka-python")
 
         # Create consumer and fetch message
         consumer = KafkaConsumer(topic, **kafka_cfg)
@@ -102,7 +102,7 @@ def fetch_metadata_confluent(bootstrap: str, cfg: Dict[str, Any]) -> Optional[Me
             "enable.auto.commit": False,
             "group.id": "superstream-metadata-fetcher",
         }
-        copy_client_configuration_properties(cfg, kafka_cfg)
+        copy_client_configuration_properties(cfg, kafka_cfg, "confluent")
 
         # Create consumer and fetch message
         consumer = Consumer(kafka_cfg)
@@ -139,7 +139,7 @@ async def fetch_metadata_aiokafka(bootstrap: str, cfg: Dict[str, Any]) -> Option
             "enable_auto_commit": False,
             "group_id": "superstream-metadata-fetcher",
         }
-        copy_client_configuration_properties(cfg, kafka_cfg)
+        copy_client_configuration_properties(cfg, kafka_cfg, "aiokafka")
 
         # Create consumer and fetch message
         consumer = AIOKafkaConsumer(topic, **kafka_cfg)

superclient/core/reporter.py

@@ -29,7 +29,7 @@ def _create_producer_kafka_python(bootstrap: str, base_cfg: Dict[str, Any]):
         "batch.size": 16_384,
         "linger.ms": 1000,
     }
-    copy_client_configuration_properties(base_cfg, cfg)
+    copy_client_configuration_properties(base_cfg, cfg, "kafka-python")
     kafka_cfg = {k.replace(".", "_"): v for k, v in cfg.items()}
     return kafka.KafkaProducer(**kafka_cfg)
 
@@ -44,7 +44,7 @@ def _create_producer_confluent(bootstrap: str, base_cfg: Dict[str, Any]):
         "batch.size": 16384,
         "linger.ms": 1000,
     }
-    copy_client_configuration_properties(base_cfg, cfg)
+    copy_client_configuration_properties(base_cfg, cfg, "confluent")
     return _CProducer(cfg)
 
 
@@ -58,7 +58,7 @@ async def _create_producer_aiokafka(bootstrap: str, base_cfg: Dict[str, Any]):
         "batch_size": 16_384,
         "linger_ms": 1000,
     }
-    copy_client_configuration_properties(base_cfg, cfg)
+    copy_client_configuration_properties(base_cfg, cfg, "aiokafka")
     return AIOKafkaProducer(**cfg)
 
 

superclient/util/config.py

@@ -52,13 +52,18 @@ def _serialize_config_value(v: Any) -> Any:
     else:
         return v
 
-def copy_client_configuration_properties(src: Dict[str, Any], dst: Dict[str, Any]):
+def copy_client_configuration_properties(src: Dict[str, Any], dst: Dict[str, Any], lib_name: str = "confluent"):
     """Copy essential client configuration properties from source to destination.
     This ensures internal Kafka clients have the same security, network, and connection
     configurations as the user's Kafka clients.
     Only copies properties that are explicitly set in the source configuration.
+    
+    Args:
+        src: Source configuration (in library-specific syntax)
+        dst: Destination configuration (in library-specific syntax)
+        lib_name: Library name for key translation
     """
-    # List of all possible auth/network related configs
+    # List of all possible auth/network related configs in Java-style syntax
     possible_keys = [
         # Security protocol
         "security.protocol",
@@ -108,10 +113,20 @@ def copy_client_configuration_properties(src: Dict[str, Any], dst: Dict[str, Any
         "retries"
     ]
 
-    # Only copy properties that are explicitly set in the source configuration
-    for k in possible_keys:
-        if k in src and k not in dst:
-            dst[k] = src[k]
+    # For each Java-style key, find if it exists in the source config
+    # by checking both the Java-style key and its library-specific equivalent
+    for java_key in possible_keys:
+        # Check if the Java-style key exists in source
+        if java_key in src:
+            if java_key not in dst:
+                dst[java_key] = src[java_key]
+            continue
+            
+        # Check if the library-specific equivalent exists in source
+        if lib_name in _JAVA_TO_LIB_MAPPING:
+            lib_key = _JAVA_TO_LIB_MAPPING[lib_name].get(java_key, java_key)
+            if lib_key in src and lib_key not in dst:
+                dst[lib_key] = src[lib_key]
 
 # ---------------------------------------------------------------------------
 # Field name mapping between Java-style and library-specific representations