support dynamic sample frequency comes as part of the metadata message

Author: idanasulin2706

.idea/runConfigurations/kafka_clients_example.xml

@@ -29,10 +29,7 @@ public class MskKafkaExample {
     private static final Logger logger = LoggerFactory.getLogger(MskKafkaExample.class);
 
     // === Configuration Constants ===
-    private static final String DEFAULT_BOOTSTRAP_SERVERS =
-            "b-23-public.superstreamstgmsk.0y88si.c2.kafka.eu-central-1.amazonaws.com:9198," +
-                    "b-24-public.superstreamstgmsk.0y88si.c2.kafka.eu-central-1.amazonaws.com:9198," +
-                    "b-2-public.superstreamstgmsk.0y88si.c2.kafka.eu-central-1.amazonaws.com:9198";
+    private static final String DEFAULT_BOOTSTRAP_SERVERS = "b-1-public.superstreamstgmsk.0y88si.c2.kafka.eu-central-1.amazonaws.com:9198,b-2-public.superstreamstgmsk.0y88si.c2.kafka.eu-central-1.amazonaws.com:9198";
 
     // AWS IAM Credentials
     private static final String AWS_ACCESS_KEY_ID = "<your-access-key>";

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

@@ -1404,10 +1404,19 @@ public static boolean markProducerClosed(Object producer) {
                 if (info.isActive.getAndSet(false)) {
                     logger.debug("Producer {} marked as closed; metrics collection will stop", producerId);
                     try {
-                        info.getReporter().deactivate();
-                    } catch (Exception ignored) {}
-                    // Remove from auxiliary map
+                        ClientStatsReporter reporter = info.getReporter();
+                        if (reporter != null) {
+                            // Stop the reporter and deregister it from coordinators
+                            reporter.deactivate();
+                            ai.superstream.core.ClientStatsReporter.deregisterReporter(reporter);
+                        }
+                    } catch (Exception e) {
+                        logger.debug("Error deactivating reporter for {}: {}", producerId, e.getMessage());
+                    }
+
+                    // Remove from lookup maps to free memory
                     clientStatsReporters.remove(producerId);
+                    producerMetricsMap.remove(producerId);
                     return true;
                 } else {
                     return false; // already closed previously

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

@@ -30,7 +30,8 @@ public class ClientStatsReporter {
     private static final SuperstreamLogger logger = SuperstreamLogger.getLogger(ClientStatsReporter.class);
     private static final String CLIENTS_TOPIC = "superstream.clients";
     private static final ObjectMapper objectMapper = new ObjectMapper();
-    private static final long REPORT_INTERVAL_MS = 300000; // 5 minutes
+    // Default reporting interval (5 minutes) – overridden when metadata provides a different value
+    private static final long DEFAULT_REPORT_INTERVAL_MS = 300000; // 5 minutes
     private static final String DISABLED_ENV_VAR = "SUPERSTREAM_DISABLED";
 
     // Shared scheduler for all reporters to minimize thread usage
@@ -282,18 +283,39 @@ private static class ClusterStatsCoordinator {
         private final CopyOnWriteArrayList<ClientStatsReporter> reporters = new CopyOnWriteArrayList<>();
         private final AtomicBoolean scheduled = new AtomicBoolean(false);
 
+        // Report interval for this cluster (milliseconds)
+        private final long reportIntervalMs;
+
         ClusterStatsCoordinator(String bootstrapServers, Properties baseProps) {
             this.bootstrapServers = bootstrapServers;
             this.baseProps = baseProps;
+
+            long interval = DEFAULT_REPORT_INTERVAL_MS;
+            try {
+                ai.superstream.model.MetadataMessage meta = ai.superstream.core.SuperstreamManager.getInstance()
+                        .getOrFetchMetadataMessage(bootstrapServers, baseProps);
+                if (meta != null && meta.getReportIntervalMs() != null && meta.getReportIntervalMs() > 0) {
+                    interval = meta.getReportIntervalMs();
+                }
+            } catch (Exception e) {
+                logger.warn("Could not obtain report interval from metadata: {}. Using default {} ms", e.getMessage(), DEFAULT_REPORT_INTERVAL_MS);
+            }
+
+            this.reportIntervalMs = interval;
         }
 
         void addReporter(ClientStatsReporter r) {
             reporters.add(r);
             if (scheduled.compareAndSet(false, true)) {
-                scheduler.scheduleAtFixedRate(this::run, REPORT_INTERVAL_MS, REPORT_INTERVAL_MS, TimeUnit.MILLISECONDS);
+                scheduler.scheduleAtFixedRate(this::run, reportIntervalMs, reportIntervalMs, TimeUnit.MILLISECONDS);
             }
         }
 
+        // Allows outer class to remove a reporter when the underlying KafkaProducer is closed
+        void removeReporter(ClientStatsReporter r) {
+            reporters.remove(r);
+        }
+
         private void run() {
             if (reporters.isEmpty())
                 return;
@@ -336,4 +358,16 @@ private void run() {
     public void deactivate() {
         registered.set(false);
     }
+
+    /**
+     * Remove the given reporter instance from all cluster coordinators. Called by the
+     * agent when the application closes its <code>KafkaProducer</code> so that we
+     * do not retain references to obsolete reporter objects.
+     */
+    public static void deregisterReporter(ClientStatsReporter reporter) {
+        if (reporter == null) {
+            return;
+        }
+        coordinators.values().forEach(coord -> coord.removeReporter(reporter));
+    }
 }

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

@@ -214,21 +214,56 @@ public boolean optimizeProducer(String bootstrapServers, String clientId, Proper
      * @return The metadata message, or null if it couldn't be retrieved
      */
     public MetadataMessage getOrFetchMetadataMessage(String bootstrapServers, Properties originalProperties) {
+        // Normalise the bootstrap servers so that different orderings of the same
+        // hosts/ports map to the *same* cache entry.  This prevents duplicate
+        // Kafka consumers and wasted network calls when the application creates
+        // multiple producers with logically-identical bootstrap lists such as
+        // "b1:9092,b2:9092" and "b2:9092,b1:9092".
+
+        String cacheKey = normalizeBootstrapServers(bootstrapServers);
+
         // Check the cache first
-        if (metadataCache.containsKey(bootstrapServers)) {
-            return metadataCache.get(bootstrapServers);
+        if (metadataCache.containsKey(cacheKey)) {
+            return metadataCache.get(cacheKey);
         }
 
-        // Fetch the metadata
+        // Fetch the metadata using the *original* string (ordering is irrelevant
+        // for the Kafka client itself)
         MetadataMessage metadataMessage = metadataConsumer.getMetadataMessage(bootstrapServers, originalProperties);
 
         if (metadataMessage != null) {
-            metadataCache.put(bootstrapServers, metadataMessage);
+            metadataCache.put(cacheKey, metadataMessage);
         }
 
         return metadataMessage;
     }
 
+    /**
+     * Produce a canonical representation of the bootstrap servers list.
+     * <p>
+     * The input may contain duplicates, whitespace or different ordering – we
+     * split on commas, trim each entry, drop empties, sort the list
+     * lexicographically and join it back with commas.  The resulting string can
+     * safely be used as a map key that uniquely identifies a Kafka cluster.
+     */
+    private static String normalizeBootstrapServers(String servers) {
+        if (servers == null) {
+            return "";
+        }
+
+        String[] parts = servers.split(",");
+        java.util.List<String> cleaned = new java.util.ArrayList<>();
+        for (String p : parts) {
+            if (p == null) continue;
+            String trimmed = p.trim();
+            if (!trimmed.isEmpty()) {
+                cleaned.add(trimmed);
+            }
+        }
+        java.util.Collections.sort(cleaned);
+        return String.join(",", cleaned);
+    }
+
     /**
      * Get the list of application topics from the environment variable.
      *

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

@@ -12,6 +12,10 @@ public class MetadataMessage {
     private boolean active;
     private List<TopicConfiguration> topicsConfiguration;
 
+    // Optional: override for client stats reporting interval (milliseconds). Can be absent (null)
+    @JsonProperty("report_interval_ms")
+    private Long reportIntervalMs;
+
     public MetadataMessage() {
         // Default constructor for Jackson
     }
@@ -50,6 +54,17 @@ public void setTopicsConfiguration(List<TopicConfiguration> topicsConfiguration)
         this.topicsConfiguration = topicsConfiguration;
     }
 
+    // Getter & setter for new report interval field
+    @JsonProperty("report_interval_ms")
+    public Long getReportIntervalMs() {
+        return reportIntervalMs;
+    }
+
+    @JsonProperty("report_interval_ms")
+    public void setReportIntervalMs(Long reportIntervalMs) {
+        this.reportIntervalMs = reportIntervalMs;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;