collect more metrics + fix loginmodule bug + fix stop collection on producer close + fix inconsistent metrics collection on idle times

Author: idanasulin2706

.idea/runConfigurations/Aiven_example.xml

@@ -29,10 +29,10 @@ public class KafkaProducerExample {
     public static void main(String[] args) {
         // Build the configuration map first using a mutable map
         Map<String, Object> mutableProps = new java.util.HashMap<>();
-        mutableProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, DEFAULT_BOOTSTRAP_SERVERS);
-//         mutableProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, Arrays.asList(DEFAULT_BOOTSTRAP_SERVERS));
+//        mutableProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, DEFAULT_BOOTSTRAP_SERVERS);
+         mutableProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, Arrays.asList(DEFAULT_BOOTSTRAP_SERVERS));
 //        mutableProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, List.of(DEFAULT_BOOTSTRAP_SERVERS));
-//        mutableProps.put(ProducerConfig.CLIENT_ID_CONFIG, CLIENT_ID);
+        mutableProps.put(ProducerConfig.CLIENT_ID_CONFIG, CLIENT_ID);
         mutableProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
         mutableProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
         mutableProps.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, COMPRESSION_TYPE);
@@ -45,6 +45,9 @@ public static void main(String[] args) {
 
         // Pass the immutable map directly to the KafkaProducer constructor
         Producer<String, String> producer = new KafkaProducer<String, String>(mutableProps);
+
+        mutableProps.put(ProducerConfig.CLIENT_ID_CONFIG, CLIENT_ID+"1");
+        Producer<String, String> producer1 = new KafkaProducer<String, String>(mutableProps);
         long recordCount = 50; // Number of messages to send
         try {
             while (true) {
@@ -53,10 +56,12 @@ public static void main(String[] args) {
                     String messageKey = MESSAGE_KEY + "-" + i;
                     String messageValue = MESSAGE_VALUE + "-" + i + "-" + System.currentTimeMillis();
                     producer.send(new ProducerRecord<>(TOPIC_NAME, messageKey, messageValue));
+                    producer.send(new ProducerRecord<>(TOPIC_NAME+"1", messageKey, messageValue));
+                    producer1.send(new ProducerRecord<>(TOPIC_NAME+"1", messageKey, messageValue));
                 }
 
                 producer.flush();
-                Thread.sleep(10000);
+                Thread.sleep(100000);
             }
         } catch (Exception e) {
             logger.error("Error sending message", e);

.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.201</version>
+  <version>1.0.202</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.201</version>
+    <version>1.0.202</version>
     <packaging>jar</packaging>
 
     <name>Superstream Kafka Client Optimizer</name>

superstream-clients/dependency-reduced-pom.xml

@@ -0,0 +1,39 @@
+package ai.superstream.agent;
+
+import ai.superstream.util.SuperstreamLogger;
+import net.bytebuddy.asm.Advice;
+
+/**
+ * Intercepts {@code KafkaProducer.close(..)} so that Superstream stops collecting
+ * metrics and sending statistics for that producer instance after the
+ * application has closed it.
+ */
+public class KafkaProducerCloseInterceptor {
+
+    // Must be non-private because instrumented KafkaProducer.close() accesses this field via generated bytecode.
+    public static final SuperstreamLogger logger = SuperstreamLogger.getLogger(KafkaProducerCloseInterceptor.class);
+
+    @Advice.OnMethodEnter
+    public static void onEnter(@Advice.This Object producer) {
+        if (KafkaProducerInterceptor.isDisabled()) {
+            return;
+        }
+
+        // Delegate to the main interceptor helper which handles thread-safety and
+        // registry updates.  We log only the first successful deactivation to
+        // avoid duplicate messages when close() delegates internally.
+        try {
+            // KafkaProducer has three close(...) overloads that delegate to each other
+            // (close(), close(Duration), close(long, TimeUnit)).  Our agent advice is
+            // woven into *all* of them, so this method is invoked once per layer of
+            // delegation.  We therefore rely on markProducerClosed() to tell us
+            // whether this is the *first* invocation for the given producer object
+            // and suppress further logging when it returns false.
+            KafkaProducerInterceptor.markProducerClosed(producer);
+            
+        } catch (Throwable ignored) {
+            // We swallow any error so that we never affect the application's own
+            // close() behaviour.
+        }
+    }
+}