report errors as metadata

Author: idanasulin2706

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

@@ -38,7 +38,7 @@ public static void main(String[] args) {
         mutableProps.put(ProducerConfig.LINGER_MS_CONFIG, 500);
 
         // Wrap the map to make it immutable – simulates a user supplying an unmodifiable configuration object
-        // Map<String, Object> props = java.util.Collections.unmodifiableMap(mutableProps);
+         Map<String, Object> props = java.util.Collections.unmodifiableMap(mutableProps);
 
         // Pass the immutable map directly to the KafkaProducer constructor
         Producer<String, String> producer = new KafkaProducer<String, String>(mutableProps);

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

@@ -2,6 +2,7 @@
 
 import ai.superstream.core.ClientStatsReporter;
 import ai.superstream.core.SuperstreamManager;
+import ai.superstream.model.MetadataMessage;
 import ai.superstream.util.SuperstreamLogger;
 import net.bytebuddy.asm.Advice;
 
@@ -16,6 +17,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.lang.ThreadLocal;
+import java.util.Collections;
 
 /**
  * Intercepts KafkaProducer constructor calls to optimize configurations and
@@ -120,6 +122,7 @@ public static void onEnter(@Advice.AllArguments Object[] args) {
             TL_UUID_STACK.get().push(producerUuid);
         }  else {
           logger.error("[ERR-002] Could not extract properties from producer arguments");
+          // here we can not report the error to the clients topic as we were not able to extract the properties
           return;
         }
 
@@ -136,19 +139,6 @@ public static void onEnter(@Advice.AllArguments Object[] args) {
             }
         }
 
-        if (immutableConfigDetected && immutableOriginalMap != null) {
-            String errMsg = String.format("[ERR-010] Cannot optimize KafkaProducer configuration: received an unmodifiable Map (%s). Please pass a mutable java.util.Properties or java.util.Map instead.",
-                    immutableOriginalMap.getClass().getName());
-            logger.error(errMsg);
-
-            // Clean up ThreadLocals so that onExit knows to skip reporter setup
-            TL_PROPS_STACK.remove();
-            TL_UUID_STACK.remove();
-
-            // Do NOT attempt optimisation
-            return;
-        }
-
         // Make a copy of the original properties in case we need to restore them
         Properties originalProperties = new Properties();
         originalProperties.putAll(properties);
@@ -179,6 +169,37 @@ public static void onEnter(@Advice.AllArguments Object[] args) {
                 return;
             }
 
+            if (immutableConfigDetected && immutableOriginalMap != null) {
+                String errMsg = String.format("[ERR-010] Cannot optimize KafkaProducer configuration: received an unmodifiable Map (%s). Please pass a mutable java.util.Properties or java.util.Map instead.",
+                        immutableOriginalMap.getClass().getName());
+                logger.error(errMsg);
+    
+                // Report the error to the client
+                try {
+                    // Get metadata message before reporting
+                    MetadataMessage metadataMessage = SuperstreamManager.getInstance().getOrFetchMetadataMessage(bootstrapServers, properties);
+                    
+                    SuperstreamManager.getInstance().reportClientInformation(
+                        bootstrapServers,
+                        properties,
+                        metadataMessage,
+                        clientId,
+                        properties,
+                        Collections.emptyMap(), // no optimized configuration since we can't optimize
+                        errMsg
+                    );
+                } catch (Exception e) {
+                    logger.error("[ERR-058] Failed to report client error: {}", e.getMessage(), e);
+                }
+    
+                // Clean up ThreadLocals so that onExit knows to skip reporter setup
+                TL_PROPS_STACK.remove();
+                TL_UUID_STACK.remove();
+    
+                // Do NOT attempt optimisation
+                return;
+            }
+
             // Optimize the producer
             boolean success = SuperstreamManager.getInstance().optimizeProducer(bootstrapServers, clientId, properties);
             if (!success) {

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

@@ -208,7 +208,7 @@ public boolean optimizeProducer(String bootstrapServers, String clientId, Proper
      * @param bootstrapServers The Kafka bootstrap servers
      * @return The metadata message, or null if it couldn't be retrieved
      */
-    private MetadataMessage getOrFetchMetadataMessage(String bootstrapServers, Properties originalProperties) {
+    public MetadataMessage getOrFetchMetadataMessage(String bootstrapServers, Properties originalProperties) {
         // Check the cache first
         if (metadataCache.containsKey(bootstrapServers)) {
             return metadataCache.get(bootstrapServers);
@@ -250,7 +250,7 @@ private List<String> getApplicationTopics() {
      * @param originalConfiguration The original configuration
      * @param optimizedConfiguration The optimized configuration
      */
-    private void reportClientInformation(String bootstrapServers, Properties originalProperties, MetadataMessage metadataMessage,
+    public void reportClientInformation(String bootstrapServers, Properties originalProperties, MetadataMessage metadataMessage,
                                          String clientId, Properties originalConfiguration,
                                          Map<String, Object> optimizedConfiguration,
                                          String error) {
@@ -269,8 +269,8 @@ private void reportClientInformation(String bootstrapServers, Properties origina
             boolean success = clientReporter.reportClient(
                     bootstrapServers,
                     originalProperties,
-                    metadataMessage.getSuperstreamClusterId(),
-                    metadataMessage.isActive(),
+                    metadataMessage != null ? metadataMessage.getSuperstreamClusterId() : null,
+                    metadataMessage != null ? metadataMessage.isActive() : false,
                     clientId,
                     originalConfiguration1,
                     optimizedConfiguration,

superstream-clients/src/main/java/ai/superstream/util/NetworkUtils.java

@@ -29,7 +29,7 @@ public static String getLocalIpAddress() {
             Enumeration<NetworkInterface> interfaces = NetworkInterface.getNetworkInterfaces();
             if (interfaces == null) {
                 logger.warn("No network interfaces found");
-                return "unknown";
+                return "";
             }
             while (interfaces.hasMoreElements()) {
                 NetworkInterface networkInterface = interfaces.nextElement();
@@ -53,7 +53,7 @@ public static String getLocalIpAddress() {
             return cachedIpAddress;
         } catch (SocketException | UnknownHostException e) {
             logger.error("[ERR-033] Failed to determine local IP address", e);
-            return "unknown";
+            return "";
         }
     }
 
@@ -73,7 +73,7 @@ public static String getHostname() {
             return cachedHostname;
         } catch (UnknownHostException e) {
             logger.error("[ERR-034] Failed to determine local hostname", e);
-            return "unknown";
+            return "";
         }
     }
 }