General enhancements

Author: idanasulin2706

.idea/runConfigurations/kafka_clients_example.xml

@@ -4,7 +4,7 @@
   <groupId>ai.superstream</groupId>
   <artifactId>superstream-clients</artifactId>
   <name>Superstream Kafka Client Optimizer</name>
-  <version>1.0.205</version>
+  <version>1.0.206</version>
   <description>A Java library that dynamically optimizes Kafka client configuration based on recommendations</description>
   <url>https://github.com/superstreamlabs/superstream-clients-java</url>
   <developers>

superstream-clients/dependency-reduced-pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>ai.superstream</groupId>
     <artifactId>superstream-clients</artifactId>
-    <version>1.0.205</version>
+    <version>1.0.206</version>
     <packaging>jar</packaging>
 
     <name>Superstream Kafka Client Optimizer</name>

superstream-clients/pom.xml

@@ -4,6 +4,7 @@
 import ai.superstream.core.SuperstreamManager;
 import ai.superstream.model.MetadataMessage;
 import ai.superstream.util.SuperstreamLogger;
+import ai.superstream.util.ClientUtils;
 import net.bytebuddy.asm.Advice;
 import java.util.AbstractMap;
 
@@ -18,7 +19,6 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.lang.ThreadLocal;
-import java.util.Collections;
 import java.util.List;
 import java.util.Arrays;
 
@@ -182,24 +182,6 @@ public static void onEnter(@Advice.AllArguments Object[] args) {
                         immutableOriginalMap.getClass().getName());
                 logger.error(errMsg);
 
-                // Report the error to the client
-                try {
-                    // Get metadata message before reporting
-                    AbstractMap.SimpleEntry<MetadataMessage, String> metadataResult = SuperstreamManager.getInstance().getOrFetchMetadataMessage(bootstrapServers, properties);
-                    MetadataMessage metadataMessage = metadataResult.getKey();
-                    SuperstreamManager.getInstance().reportClientInformation(
-                        bootstrapServers,
-                        properties,
-                        metadataMessage,
-                        clientId,
-                        properties,
-                        Collections.emptyMap(), // no optimized configuration since we can't optimize
-                        errMsg
-                    );
-                } catch (Exception e) {
-                    logger.error("[ERR-058] Failed to report client error: {}", e.getMessage(), e);
-                }
-
                 // Push ConfigInfo with error and original config for stats reporting
                 java.util.Deque<ConfigInfo> cfgStack = TL_CFG_STACK.get();
                 cfgStack.push(new ConfigInfo(propertiesToMap(properties), new java.util.HashMap<>(), errMsg));
@@ -349,7 +331,7 @@ public static void onExit(@Advice.This Object producer) {
                 }
                 if (cfgInfo != null) {
                     // Use the original configuration from ConfigInfo and get complete config with defaults
-                    java.util.Map<String, Object> completeConfig = ai.superstream.core.ClientReporter.getCompleteProducerConfig(cfgInfo.originalConfig);
+                    java.util.Map<String, Object> completeConfig = ClientUtils.getCompleteProducerConfig(cfgInfo.originalConfig);
                     java.util.Map<String, Object> optimizedConfig = cfgInfo.optimizedConfig != null ? cfgInfo.optimizedConfig : new java.util.HashMap<>();
                     reporter.setConfigurations(completeConfig, optimizedConfig);
                     // If optimizedConfig is empty and there is an error, set the error on the reporter
@@ -360,7 +342,7 @@ public static void onExit(@Advice.This Object producer) {
                     // No ConfigInfo available, so no optimization was performed
                     // Use the producer properties as both original and optimized (since no changes were made)
                     java.util.Map<String, Object> originalPropsMap = propertiesToMap(producerProps);
-                    java.util.Map<String, Object> completeConfig = ai.superstream.core.ClientReporter.getCompleteProducerConfig(originalPropsMap);
+                    java.util.Map<String, Object> completeConfig = ClientUtils.getCompleteProducerConfig(originalPropsMap);
                     reporter.setConfigurations(completeConfig, new java.util.HashMap<>());
                 }
 

superstream-clients/src/main/java/ai/superstream/agent/KafkaProducerInterceptor.java

@@ -5,6 +5,7 @@
 import ai.superstream.util.NetworkUtils;
 import ai.superstream.util.SuperstreamLogger;
 import ai.superstream.util.KafkaPropertiesUtils;
+import ai.superstream.util.ClientUtils;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.kafka.clients.producer.KafkaProducer;
@@ -139,7 +140,7 @@ void drainInto(Producer<String, String> producer) {
                     NetworkUtils.getLocalIpAddress(),
                     totalBytesBefore,
                     totalBytesAfter,
-                    ClientReporter.getClientVersion(),
+                    ClientUtils.getClientVersion(),
                     NetworkUtils.getHostname(),
                     producerUuid);
 
@@ -336,7 +337,8 @@ private void run() {
                 baseProps.forEach((key, value) -> {
                     // Mask sensitive values
                     if (key.toString().toLowerCase().contains("password") || 
-                        key.toString().toLowerCase().contains("sasl.jaas.config")) {
+                        key.toString().toLowerCase().contains("sasl.jaas.config") ||
+                        key.toString().equals("basic.auth.user.info")) {
                         configLog.append(key).append("=[MASKED], ");
                     } else {
                         configLog.append(key).append("=").append(value).append(", ");

superstream-clients/src/main/java/ai/superstream/core/ClientReporter.java

@@ -50,7 +50,8 @@ public java.util.AbstractMap.SimpleEntry<MetadataMessage, String> getMetadataMes
             properties.forEach((key, value) -> {
                 // Mask sensitive values
                 if (key.toString().toLowerCase().contains("password") || 
-                    key.toString().toLowerCase().contains("sasl.jaas.config")) {
+                    key.toString().toLowerCase().contains("sasl.jaas.config") ||
+                    key.toString().equals("basic.auth.user.info")) {
                     configLog.append(key).append("=[MASKED], ");
                 } else {
                     configLog.append(key).append("=").append(value).append(", ");