populate stats messages with error that was missed

Author: idanasulin2706

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

@@ -18,6 +18,8 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.lang.ThreadLocal;
 import java.util.Collections;
+import java.util.List;
+import java.util.Arrays;
 
 /**
  * Intercepts KafkaProducer constructor calls to optimize configurations and
@@ -206,7 +208,10 @@ public static void onEnter(@Advice.AllArguments Object[] args) {
                     properties.putAll(originalProperties);
                     // Push ConfigInfo with original config for stats reporting
                     java.util.Deque<ConfigInfo> cfgStack = TL_CFG_STACK.get();
-                    cfgStack.push(new ConfigInfo(originalPropertiesMap, new java.util.HashMap<>()));
+                    // Check if there's an existing ConfigInfo with an error
+                    ConfigInfo existingConfig = cfgStack.isEmpty() ? null : cfgStack.peek();
+                    String error = existingConfig != null ? existingConfig.error : null;
+                    cfgStack.push(new ConfigInfo(originalPropertiesMap, new java.util.HashMap<>(), error));
                 }
             } catch (Exception e) {
                 logger.error("[ERR-053] Error during producer optimization: {}", e.getMessage(), e);
@@ -291,21 +296,21 @@ public static void onExit(@Advice.This Object producer) {
 
                 // Set the most impactful topic if possible
                 try {
-                    ai.superstream.model.MetadataMessage metadataMessage = null;
-                    java.util.List<String> topics = null;
+                    MetadataMessage metadataMessage = null;
+                    List<String> topics = null;
                     // Try to get metadata and topics if available
                     if (producerProps != null) {
                         String bootstrapServersProp = producerProps.getProperty("bootstrap.servers");
                         if (bootstrapServersProp != null) {
-                            metadataMessage = ai.superstream.core.SuperstreamManager.getInstance().getOrFetchMetadataMessage(bootstrapServersProp, producerProps);
+                            metadataMessage = SuperstreamManager.getInstance().getOrFetchMetadataMessage(bootstrapServersProp, producerProps);
                         }
                         String topicsEnv = System.getenv("SUPERSTREAM_TOPICS_LIST");
                         if (topicsEnv != null && !topicsEnv.trim().isEmpty()) {
-                            topics = java.util.Arrays.asList(topicsEnv.split(","));
+                            topics = Arrays.asList(topicsEnv.split(","));
                         }
                     }
                     if (metadataMessage != null && topics != null) {
-                        String mostImpactfulTopic = ai.superstream.core.SuperstreamManager.getInstance().getConfigurationOptimizer().getMostImpactfulTopicName(metadataMessage, topics);
+                        String mostImpactfulTopic = SuperstreamManager.getInstance().getConfigurationOptimizer().getMostImpactfulTopicName(metadataMessage, topics);
                         if (mostImpactfulTopic == null) {
                             mostImpactfulTopic = "";
                         }

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

@@ -128,6 +128,10 @@ public boolean optimizeProducer(String bootstrapServers, String clientId, Proper
                 String errMsg = "[ERR-054] Superstream optimization is not active for this kafka cluster, please head to the Superstream console and activate it.";
                 logger.error(errMsg);
                 reportClientInformation(bootstrapServers, properties, metadataMessage, clientId, originalProperties, Collections.emptyMap(), errMsg);
+                
+                // Push ConfigInfo with error and original config for stats reporting
+                java.util.Deque<ai.superstream.agent.KafkaProducerInterceptor.ConfigInfo> cfgStack = ai.superstream.agent.KafkaProducerInterceptor.TL_CFG_STACK.get();
+                cfgStack.push(new ai.superstream.agent.KafkaProducerInterceptor.ConfigInfo(convertPropertiesToMap(properties), new java.util.HashMap<>(), errMsg));
                 return false;
             }