report error as part of the metadata

Author: idanasulin2706

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

@@ -40,12 +40,14 @@ public class ClientReporter {
      * @param optimizedConfiguration The optimized configuration
      * @param mostImpactfulTopic The most impactful topic
      * @param producerUuid The producer UUID to include in the report
+     * @param error The error string to include in the report
      * @return True if the message was sent successfully, false otherwise
      */
     public boolean reportClient(String bootstrapServers, Properties originalClientProperties, int superstreamClusterId, boolean active,
                                 String clientId, Map<String, Object> originalConfiguration,
                                 Map<String, Object> optimizedConfiguration,
-                                String mostImpactfulTopic, String producerUuid) {
+                                String mostImpactfulTopic, String producerUuid,
+                                String error) {
         Properties properties = new Properties();
 
         // Copy all authentication-related and essential properties from the original client
@@ -74,7 +76,8 @@ public boolean reportClient(String bootstrapServers, Properties originalClientPr
                     optimizedConfiguration,
                     mostImpactfulTopic,
                     NetworkUtils.getHostname(),
-                    producerUuid
+                    producerUuid,
+                    error
             );
 
             // Convert the message to JSON

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

@@ -116,8 +116,9 @@ public boolean optimizeProducer(String bootstrapServers, String clientId, Proper
 
             // Check if optimization is active
             if (!metadataMessage.isActive()) {
-                logger.error("[ERR-054] Superstream optimization is not active for this kafka cluster, please head to the Superstream console and activate it.");
-                reportClientInformation(bootstrapServers, properties, metadataMessage, clientId, originalProperties, Collections.emptyMap());
+                String errMsg = "[ERR-054] Superstream optimization is not active for this kafka cluster, please head to the Superstream console and activate it.";
+                logger.error(errMsg);
+                reportClientInformation(bootstrapServers, properties, metadataMessage, clientId, originalProperties, Collections.emptyMap(), errMsg);
                 return false;
             }
 
@@ -133,7 +134,7 @@ public boolean optimizeProducer(String bootstrapServers, String clientId, Proper
 
             if (modifiedKeys.isEmpty()) {
                 logger.debug("No configuration parameters were modified");
-                reportClientInformation(bootstrapServers, properties, metadataMessage, clientId, originalProperties, Collections.emptyMap());
+                reportClientInformation(bootstrapServers, properties, metadataMessage, clientId, originalProperties, Collections.emptyMap(), "");
                 return false;
             }
 
@@ -180,7 +181,8 @@ public boolean optimizeProducer(String bootstrapServers, String clientId, Proper
                     metadataMessage,
                     clientId,
                     originalProperties,
-                    optimizedProperties
+                    optimizedProperties,
+                    ""
             );
 
             // Log optimization success with linger.ms status based on environment variable
@@ -250,7 +252,8 @@ private List<String> getApplicationTopics() {
      */
     private void reportClientInformation(String bootstrapServers, Properties originalProperties, MetadataMessage metadataMessage,
                                          String clientId, Properties originalConfiguration,
-                                         Map<String, Object> optimizedConfiguration) {
+                                         Map<String, Object> optimizedConfiguration,
+                                         String error) {
         try {
             Map<String, Object> originalConfiguration1 = convertPropertiesToMap(originalConfiguration);
             List<String> topics = getApplicationTopics();
@@ -272,7 +275,8 @@ private void reportClientInformation(String bootstrapServers, Properties origina
                     originalConfiguration1,
                     optimizedConfiguration,
                     mostImpactfulTopic,
-                    producerUuid
+                    producerUuid,
+                    error
             );
 
             if (!success) {

superstream-clients/src/main/java/ai/superstream/model/ClientMessage.java

@@ -30,7 +30,7 @@ public ClientMessage() {
 
     public ClientMessage(int superstreamClusterId, boolean active, String clientId, String ipAddress, String clientVersion, String language, String clientType,
                          Map<String, Object> originalConfiguration, Map<String, Object> optimizedConfiguration,
-                         String mostImpactfulTopic, String hostname, String producerUuid) {
+                         String mostImpactfulTopic, String hostname, String producerUuid, String error) {
         this.superstreamClusterId = superstreamClusterId;
         this.active = active;
         this.clientId = clientId;
@@ -44,6 +44,7 @@ public ClientMessage(int superstreamClusterId, boolean active, String clientId,
         this.environmentVariables = ai.superstream.util.EnvironmentVariables.getSuperstreamEnvironmentVariables();
         this.hostname = hostname;
         this.producerUuid = producerUuid;
+        this.error = error;
     }
 
     @JsonProperty("superstream_cluster_id")