add immidiate stats report after producer creation

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

superstream-clients/pom.xml

@@ -345,6 +345,13 @@ public static void onExit(@Advice.This Object producer) {
                     reporter.setConfigurations(completeConfig, new java.util.HashMap<>());
                 }
 
+                // Trigger immediate metrics collection for this producer
+                try {
+                    sharedCollector.collectMetricsForProducer(producerId, metricsInfo);
+                } catch (Exception e) {
+                    logger.error("[ERR-047] Failed to collect immediate metrics for new producer {}: {}", producerId, e.getMessage(), e);
+                }
+
                 logger.debug("Producer {} registered with shared metrics collector", producerId);
             }
         } catch (Exception e) {

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

@@ -160,10 +160,8 @@ void drainInto(Producer<String, String> producer) {
             message.setOriginalConfiguration(originalConfig != null ? originalConfig : new java.util.HashMap<>());
             message.setOptimizedConfiguration(optimizedConfig != null ? optimizedConfig : new java.util.HashMap<>());
 
-            // Attach topics list
-            if (!topicsWritten.isEmpty()) {
-                message.setTopics(new java.util.ArrayList<>(topicsWritten));
-            }
+            // Attach topics list - always set it, empty if no topics
+            message.setTopics(new java.util.ArrayList<>(topicsWritten));
 
             // When building the ClientStatsMessage, set the most impactful topic if available
             if (mostImpactfulTopic != null) {
@@ -178,7 +176,6 @@ void drainInto(Producer<String, String> producer) {
             ProducerRecord<String, String> record = new ProducerRecord<>(CLIENTS_TOPIC, json);
             producer.send(record);
 
-            // Log at INFO level that stats have been sent for this producer
             logger.debug("Producer {} stats sent: before={} bytes, after={} bytes",
                     clientId, totalBytesBefore, totalBytesAfter);
         } catch (Exception e) {
@@ -308,6 +305,17 @@ private static class ClusterStatsCoordinator {
 
         void addReporter(ClientStatsReporter r) {
             reporters.add(r);
+            // Schedule immediate run for this specific reporter
+            scheduler.schedule(() -> {
+                try (Producer<String, String> producer = new KafkaProducer<>(baseProps)) {
+                    r.drainInto(producer);
+                    producer.flush();
+                } catch (Exception e) {
+                    logger.error("[ERR-046] Failed to send immediate stats for new reporter: {}", e.getMessage(), e);
+                }
+            }, 0, TimeUnit.MILLISECONDS);
+
+            // Only schedule the periodic task if not already scheduled
             if (scheduled.compareAndSet(false, true)) {
                 scheduler.scheduleAtFixedRate(this::run, reportIntervalMs, reportIntervalMs, TimeUnit.MILLISECONDS);
             }