logs improvements + add metadata

Author: idanasulin2706

.idea/runConfigurations/kafka_clients_example.xml

@@ -39,7 +39,8 @@ public static void main(String[] args) {
 
 
         long recordCount = 50; // Number of messages to send
-        try (Producer<String, String> producer = new KafkaProducer<>(props)) {
+        Producer<String, String> producer = new KafkaProducer<>(props);
+        try {
             while (true) {
                 // Send 50 large messages to see compression benefits
                 for (int i = 1; i <= recordCount; i++) {
@@ -49,10 +50,12 @@ public static void main(String[] args) {
                 }
 
                 producer.flush();
-                Thread.sleep(100000);
+                Thread.sleep(10000);
             }
         } catch (Exception e) {
             logger.error("Error sending message", e);
+        } finally {
+            producer.close();
         }
     }
 

.idea/runConfigurations/spring_example.xml

@@ -4,7 +4,7 @@
   <groupId>ai.superstream</groupId>
   <artifactId>superstream-clients</artifactId>
   <name>Superstream Kafka Client Optimizer</name>
-  <version>1.0.16</version>
+  <version>1.0.17</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>

examples/kafka-clients-example/src/main/java/ai/superstream/examples/KafkaProducerExample.java

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

superstream-clients/dependency-reduced-pom.xml

@@ -9,7 +9,6 @@
 import java.lang.reflect.Method;
 import java.util.Properties;
 import java.util.Map;
-import java.util.List;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.Executors;
@@ -48,6 +47,13 @@ public class KafkaProducerInterceptor {
     public static final ThreadLocal<java.util.Deque<Properties>> TL_PROPS_STACK =
             ThreadLocal.withInitial(java.util.ArrayDeque::new);
 
+    // ThreadLocal stack to hold the producer UUIDs generated in onEnter so that the same
+    // value can be reused later in onExit (for stats reporting) and by SuperstreamManager
+    // when it reports the client information. The stack is aligned with the TL_PROPS_STACK
+    // (push in onEnter, pop in onExit).
+    public static final ThreadLocal<java.util.Deque<String>> TL_UUID_STACK =
+            ThreadLocal.withInitial(java.util.ArrayDeque::new);
+
     // ThreadLocal stack to pass original/optimized configuration maps from optimization phase to reporter creation.
     public static final ThreadLocal<java.util.Deque<ConfigInfo>> TL_CFG_STACK =
             ThreadLocal.withInitial(java.util.ArrayDeque::new);
@@ -57,12 +63,11 @@ public class KafkaProducerInterceptor {
         if (!DISABLED) {
             try {
                 sharedCollector.start();
-                logger.info("Superstream metrics collector initialized and started successfully");
             } catch (Exception e) {
-                logger.error("Failed to start metrics collector: " + e.getMessage(), e);
+                logger.error("[ERR-001] Failed to start metrics collector: {}", e.getMessage(), e);
             }
         } else {
-            logger.info("Superstream metrics collection is disabled via SUPERSTREAM_DISABLED environment variable");
+            logger.warn("Superstream is disabled via SUPERSTREAM_DISABLED environment variable");
         }
     }
 
@@ -109,8 +114,12 @@ public static void onEnter(@Advice.AllArguments Object[] args) {
             }
 
             TL_PROPS_STACK.get().push(properties);
+
+            // Generate a UUID for this upcoming producer instance and push onto UUID stack
+            String producerUuid = java.util.UUID.randomUUID().toString();
+            TL_UUID_STACK.get().push(producerUuid);
         }  else {
-          logger.error("Could not extract properties from producer arguments");
+          logger.error("[ERR-002] Could not extract properties from producer arguments");
           return;
         }
 
@@ -126,7 +135,7 @@ public static void onEnter(@Advice.AllArguments Object[] args) {
             }
 
             if (properties == null || properties.isEmpty()) {
-                logger.error("Could not extract properties from properties");
+                logger.error("[ERR-003] Could not extract properties from properties");
                 return;
             }
 
@@ -137,36 +146,21 @@ public static void onEnter(@Advice.AllArguments Object[] args) {
                 return;
             }
 
-            logger.info("Intercepted KafkaProducer constructor");
-
             // Extract bootstrap servers
             String bootstrapServers = properties.getProperty("bootstrap.servers");
             if (bootstrapServers == null || bootstrapServers.trim().isEmpty()) {
-                logger.warn("bootstrap.servers is not set, cannot optimize");
+                logger.error("[ERR-004] bootstrap.servers is not set, cannot optimize");
                 return;
             }
 
             // Optimize the producer
             boolean success = SuperstreamManager.getInstance().optimizeProducer(bootstrapServers, clientId, properties);
             if (!success) {
-                /*
-                 * Roll-back any changes we may have applied without wiping the whole
-                 * Properties object first.  Clearing the map created a tiny window in
-                 * which application-code that already held the reference could observe
-                 * an empty map – leading to NPEs further down the stack.  Instead we
-                 * simply overlay the original values which is functionally equivalent
-                 * but keeps the map non-empty at all times.
-                 */
                 properties.putAll(originalProperties);
             }
         } catch (Exception e) {
-            /*
-             * If optimisation threw, restore the original entries but keep the map
-             * non-empty so that any other thread holding a reference never sees an
-             * empty state.
-             */
             properties.putAll(originalProperties);
-            logger.error("Error during producer optimization, restored original properties", e);
+            logger.error("[ERR-005] Error during producer optimization, restored original properties: {}", e.getMessage(), e);
         }
     }
 
@@ -192,20 +186,30 @@ public static void onExit(@Advice.This Object producer) {
 
             java.util.Deque<Properties> stack = TL_PROPS_STACK.get();
             if (stack.isEmpty()) {
-                logger.error("No captured properties for this producer constructor; skipping stats reporter setup");
+                logger.error("[ERR-006] No captured properties for this producer constructor; skipping stats reporter setup");
                 return;
             }
 
             Properties producerProps = stack.pop();
 
+            // Retrieve matching UUID for this constructor instance
+            java.util.Deque<String> uuidStack = TL_UUID_STACK.get();
+            String producerUuid = "";
+            if (!uuidStack.isEmpty()) {
+                producerUuid = uuidStack.pop();
+            } else {
+                logger.error("[ERR-127] No producer UUID found for this constructor instance");
+            }
+
             // Clean up ThreadLocal when outer-most constructor finishes
             if (stack.isEmpty()) {
                 TL_PROPS_STACK.remove();
+                TL_UUID_STACK.remove();
             }
 
             String bootstrapServers = producerProps.getProperty("bootstrap.servers");
             if (bootstrapServers == null || bootstrapServers.isEmpty()) {
-                logger.error("bootstrap.servers missing in captured properties; skipping reporter setup");
+                logger.error("[ERR-007] bootstrap.servers missing in captured properties; skipping reporter setup");
                 return;
             }
 
@@ -228,29 +232,29 @@ public static void onExit(@Advice.This Object producer) {
                 logger.debug("Registering producer with metrics collector: {} (client.id='{}')", producerId, clientIdForStats);
 
                 // Create a reporter for this producer instance – pass the original client.id
-                ClientStatsReporter reporter = new ClientStatsReporter(bootstrapServers, producerProps, clientIdForStats);
+                ClientStatsReporter reporter = new ClientStatsReporter(bootstrapServers, producerProps, clientIdForStats, producerUuid);
 
-                // Create metrics info for this producer
-                ProducerMetricsInfo metricsInfo = new ProducerMetricsInfo(producer, reporter);
+                    // Create metrics info for this producer
+                    ProducerMetricsInfo metricsInfo = new ProducerMetricsInfo(producer, reporter);
 
-                // Register with the shared collector
-                producerMetricsMap.put(producerId, metricsInfo);
-                clientStatsReporters.put(producerId, reporter);
+                    // Register with the shared collector
+                    producerMetricsMap.put(producerId, metricsInfo);
+                    clientStatsReporters.put(producerId, reporter);
 
-                // Pop configuration info from ThreadLocal stack (if any) and attach to reporter
-                java.util.Deque<ConfigInfo> cfgStack = TL_CFG_STACK.get();
-                ConfigInfo cfgInfo = cfgStack.isEmpty()? null : cfgStack.pop();
-                if (cfgStack.isEmpty()) {
-                    TL_CFG_STACK.remove();
-                }
-                if (cfgInfo != null) {
-                    reporter.setConfigurations(cfgInfo.originalConfig, cfgInfo.optimizedConfig);
-                }
+                    // Pop configuration info from ThreadLocal stack (if any) and attach to reporter
+                    java.util.Deque<ConfigInfo> cfgStack = TL_CFG_STACK.get();
+                    ConfigInfo cfgInfo = cfgStack.isEmpty()? null : cfgStack.pop();
+                    if (cfgStack.isEmpty()) {
+                        TL_CFG_STACK.remove();
+                    }
+                    if (cfgInfo != null) {
+                        reporter.setConfigurations(cfgInfo.originalConfig, cfgInfo.optimizedConfig);
+                    }
 
                 logger.debug("Producer {} registered with shared metrics collector", producerId);
             }
         } catch (Exception e) {
-            logger.error("Error registering producer with metrics collector: " + e.getMessage(), e);
+            logger.error("[ERR-008] Error registering producer with metrics collector: {}", e.getMessage(), e);
         }
     }
 
@@ -260,7 +264,7 @@ public static void onExit(@Advice.This Object producer) {
     public static Properties extractProperties(Object[] args) {
         // Look for Properties or Map in the arguments
         if (args == null) {
-            logger.error("extractProperties: args array is null");
+            logger.error("[ERR-009] extractProperties: args array is null");
             return null;
         }
 
@@ -284,7 +288,7 @@ public static Properties extractProperties(Object[] args) {
 
                 // If the map is unmodifiable we cannot change producer configuration -> cannot optimise
                 if (arg.getClass().getName().contains("UnmodifiableMap")) {
-                    logger.error("Cannot optimize KafkaProducer configuration: received an unmodifiable Map ({}). " +
+                    logger.error("[ERR-010] Cannot optimize KafkaProducer configuration: received an unmodifiable Map ({}). " +
                                  "Please pass a mutable java.util.Properties or java.util.Map instead.",
                                  arg.getClass().getName());
                     return null; // signal caller to skip optimisation
@@ -298,7 +302,7 @@ public static Properties extractProperties(Object[] args) {
                     return props;
                 } catch (ClassCastException e) {
                     // Not the map type we expected
-                    logger.error("extractProperties: Could not cast Map to Map<String, Object>", e);
+                    logger.error("[ERR-011] extractProperties: Could not cast Map to Map<String, Object>: {}", e.getMessage(), e);
                     return null;
                 }
             }
@@ -334,7 +338,7 @@ public static Properties extractProperties(Object[] args) {
                             }
                         } catch (NoSuchFieldException e) {
                             // Field doesn't exist, try the next one
-                            logger.error("extractProperties: Field {} not found in ProducerConfig", fieldName);
+                            logger.error("[ERR-017] extractProperties: Field {} not found in ProducerConfig", fieldName);
                             continue;
                         }
                     }
@@ -399,13 +403,14 @@ public static Properties extractProperties(Object[] args) {
                     }
 
                 } catch (Exception e) {
-                    logger.error("extractProperties: Failed to extract properties from ProducerConfig: {}",
+                    logger.error("[ERR-018] extractProperties: Failed to extract properties from ProducerConfig: {}",
                             e.getMessage(), e);
+                            return null;
                 }
             }
         }
 
-        logger.error("extractProperties: No valid configuration object found in arguments");
+        logger.error("[ERR-019] extractProperties: No valid configuration object found in arguments");
         return null;
     }
 
@@ -595,13 +600,12 @@ public void start() {
                             COLLECTION_INTERVAL_MS / 2, // Start sooner for first collection
                             COLLECTION_INTERVAL_MS,
                             TimeUnit.MILLISECONDS);
-                    logger.debug("Metrics collection scheduler started successfully");
                 } catch (Exception e) {
-                    logger.error("Failed to schedule metrics collection: " + e.getMessage(), e);
+                    logger.error("[ERR-012] Failed to schedule metrics collection: {}", e.getMessage(), e);
                     running.set(false);
                 }
             } else {
-                logger.info("Metrics collector already running");
+                logger.debug("Metrics collector already running");
             }
         }
 
@@ -644,7 +648,7 @@ public void collectAllMetrics() {
                             skippedCount++;
                         }
                     } catch (Exception e) {
-                        logger.error("Error collecting metrics for producer {}: {}", producerId, e.getMessage());
+                        logger.error("[ERR-013] Error collecting metrics for producer {}: {}", producerId, e.getMessage(), e);
                     }
                 }
 
@@ -653,7 +657,7 @@ public void collectAllMetrics() {
                         totalProducers, successCount, skippedCount);
 
             } catch (Exception e) {
-                logger.error("Error in metrics collection cycle: " + e.getMessage(), e);
+                logger.error("[ERR-014] Error in metrics collection cycle: {}", e.getMessage(), e);
             }
         }
 
@@ -757,7 +761,7 @@ public boolean collectMetricsForProducer(String producerId, ProducerMetricsInfo
                         }
                     }
                 } catch (Exception snapshotEx) {
-                    // ignore snapshot errors
+                    logger.error("[ERR-015] Error extracting metrics snapshot for producer {}: {}", producerId, snapshotEx.getMessage(), snapshotEx);
                 }
 
                 // Update reporter with latest metrics snapshot
@@ -805,7 +809,7 @@ public boolean collectMetricsForProducer(String producerId, ProducerMetricsInfo
 
                 return true;
             } catch (Exception e) {
-                logger.error("Error collecting Kafka metrics for producer {}: {}", producerId, e.getMessage(), e);
+                logger.error("[ERR-016] Error collecting Kafka metrics for producer {}: {}", producerId, e.getMessage(), e);
                 return false;
             }
         }