reporting error in case of error while fetching the metadata message

Author: idanasulin2706

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

@@ -292,7 +292,7 @@ private static class ClusterStatsCoordinator {
             long interval = DEFAULT_REPORT_INTERVAL_MS;
             try {
                 ai.superstream.model.MetadataMessage meta = ai.superstream.core.SuperstreamManager.getInstance()
-                        .getOrFetchMetadataMessage(bootstrapServers, baseProps);
+                        .getOrFetchMetadataMessage(bootstrapServers, baseProps).getKey();
                 if (meta != null && meta.getReportIntervalMs() != null && meta.getReportIntervalMs() > 0) {
                     interval = meta.getReportIntervalMs();
                 }

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

@@ -30,9 +30,9 @@ public class MetadataConsumer {
      * Get the metadata message from the Kafka cluster.
      *
      * @param bootstrapServers The Kafka bootstrap servers
-     * @return The metadata message, or null if there was an error
+     * @return A pair containing the metadata message (or null if error) and the error message (or null if no error)
      */
-    public MetadataMessage getMetadataMessage(String bootstrapServers, Properties originalClientProperties) {
+    public java.util.AbstractMap.SimpleEntry<MetadataMessage, String> getMetadataMessage(String bootstrapServers, Properties originalClientProperties) {
         Properties properties = new Properties();
 
         // Copy essential client configuration properties from the original client
@@ -67,8 +67,9 @@ public MetadataMessage getMetadataMessage(String bootstrapServers, Properties or
             // Check if the metadata topic exists
             Set<String> topics = consumer.listTopics().keySet();
             if (!topics.contains(METADATA_TOPIC)) {
-                logger.error("[ERR-034] Superstream internal topic is missing. This topic is required for Superstream to function properly. Please make sure the Kafka user has read/write/describe permissions on superstream.* topics.");
-                return null;
+                String errMsg = "[ERR-034] Superstream internal topic is missing. This topic is required for Superstream to function properly. Please make sure the Kafka user has read/write/describe permissions on superstream.* topics.";
+                logger.error(errMsg);
+                return new java.util.AbstractMap.SimpleEntry<>(null, errMsg);
             }
 
             // Assign the metadata topic
@@ -80,8 +81,9 @@ public MetadataMessage getMetadataMessage(String bootstrapServers, Properties or
             long endOffset = consumer.position(partition);
 
             if (endOffset == 0) {
-                logger.error("[ERR-035] Unable to retrieve optimizations data from Superstream. This is required for optimization. Please contact the Superstream team if the issue persists.");
-                return null;
+                String errMsg = "[ERR-035] Unable to retrieve optimizations data from Superstream. This is required for optimization. Please contact the Superstream team if the issue persists.";
+                logger.error(errMsg);
+                return new java.util.AbstractMap.SimpleEntry<>(null, errMsg);
             }
 
             // Seek to the last message
@@ -90,20 +92,23 @@ public MetadataMessage getMetadataMessage(String bootstrapServers, Properties or
             // Poll for the message
             ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(5));
             if (records.isEmpty()) {
-                logger.error("[ERR-036] Unable to retrieve optimizations data from Superstream. This is required for optimization. Please contact the Superstream team if the issue persists.");
-                return null;
+                String errMsg = "[ERR-036] Unable to retrieve optimizations data from Superstream. This is required for optimization. Please contact the Superstream team if the issue persists.";
+                logger.error(errMsg);
+                return new java.util.AbstractMap.SimpleEntry<>(null, errMsg);
             }
             logger.debug("Successfully retrieved a message from the {} topic", METADATA_TOPIC);
 
             // Parse the message
             String json = records.iterator().next().value();
-            return objectMapper.readValue(json, MetadataMessage.class);
+            return new java.util.AbstractMap.SimpleEntry<>(objectMapper.readValue(json, MetadataMessage.class), null);
         } catch (IOException e) {
-            logger.error("[ERR-027] Unable to retrieve optimizations data from Superstream. This is required for optimization. Please contact the Superstream team if the issue persists: {}", e.getMessage(), e);
-            return null;
+            String errMsg = "[ERR-027] Unable to retrieve optimizations data from Superstream. This is required for optimization. Please contact the Superstream team if the issue persists: " + e.getMessage();
+            logger.error(errMsg, e);
+            return new java.util.AbstractMap.SimpleEntry<>(null, errMsg);
         } catch (Exception e) {
-            logger.error("[ERR-028] Unable to retrieve optimizations data from Superstream. This is required for optimization. Please make sure the Kafka user has read/write/describe permissions on superstream.* topics: {}", e.getMessage(), e);
-            return null;
+            String errMsg = "[ERR-028] Unable to retrieve optimizations data from Superstream. This is required for optimization. Please make sure the Kafka user has read/write/describe permissions on superstream.* topics: " + e.getMessage();
+            logger.error(errMsg, e);
+            return new java.util.AbstractMap.SimpleEntry<>(null, errMsg);
         }
     }
 }

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

@@ -113,9 +113,15 @@ public boolean optimizeProducer(String bootstrapServers, String clientId, Proper
             setOptimizationInProgress(true);
 
             // Get or fetch the metadata message
-            MetadataMessage metadataMessage = getOrFetchMetadataMessage(bootstrapServers, properties);
+            java.util.AbstractMap.SimpleEntry<MetadataMessage, String> result = getOrFetchMetadataMessage(bootstrapServers, properties);
+            MetadataMessage metadataMessage = result.getKey();
+            String error = result.getValue();
+
             if (metadataMessage == null) {
-                // log is inside the getOrFetchMetadataMessage method
+                // Error is already logged in getOrFetchMetadataMessage
+                // 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<>(), error));
                 return false;
             }
 
@@ -215,11 +221,10 @@ public boolean optimizeProducer(String bootstrapServers, String clientId, Proper
      * Get the metadata message for a given Kafka cluster.
      *
      * @param bootstrapServers The Kafka bootstrap servers
-     * @return The metadata message, or null if it couldn't be retrieved
+     * @return A pair containing the metadata message (or null if error) and the error message (or null if no error)
      */
-    public MetadataMessage getOrFetchMetadataMessage(String bootstrapServers, Properties originalProperties) {
+    public java.util.AbstractMap.SimpleEntry<MetadataMessage, String> 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".
@@ -228,18 +233,19 @@ public MetadataMessage getOrFetchMetadataMessage(String bootstrapServers, Proper
 
         // Check the cache first
         if (metadataCache.containsKey(cacheKey)) {
-            return metadataCache.get(cacheKey);
+            return new java.util.AbstractMap.SimpleEntry<>(metadataCache.get(cacheKey), null);
         }
 
         // Fetch the metadata using the *original* string (ordering is irrelevant
         // for the Kafka client itself)
-        MetadataMessage metadataMessage = metadataConsumer.getMetadataMessage(bootstrapServers, originalProperties);
+        java.util.AbstractMap.SimpleEntry<MetadataMessage, String> result = metadataConsumer.getMetadataMessage(bootstrapServers, originalProperties);
+        MetadataMessage metadataMessage = result.getKey();
 
         if (metadataMessage != null) {
             metadataCache.put(cacheKey, metadataMessage);
         }
 
-        return metadataMessage;
+        return result;
     }
 
     /**