#211 Feature producer header flag fix (#223)

* header config changes

conf/configuration.yml

@@ -81,3 +81,17 @@ otelOptions:
   otlp.aggregationTemporality: "CUMULATIVE"
   otlp.resourceAttributes:
   otlp.headers:
+
+messageHeaderConfiguration:
+  allowedPrefix: ["X_"]
+  msgIdHeader: "X_MESSAGE_ID"
+  groupIdHeader: "X_GROUP_ID"
+  callbackCodes: "X_CALLBACK_CODES"
+  requestTimeout: "X_REQUEST_TIMEOUT"
+  replyToHttpUriHeader: "X_REPLY_TO_HTTP_URI"
+  replyToHttpMethodHeader: "X_REPLY_TO_HTTP_METHOD"
+  replyToHeader: "X_REPLY_TO"
+  httpUriHeader: "X_HTTP_URI"
+  httpMethodHeader: "X_HTTP_METHOD"
+  httpContentType: "X_CONTENT_TYPE"
+

server/src/main/java/com/flipkart/varadhi/VaradhiApplication.java

@@ -150,7 +150,9 @@ public static AppConfiguration readConfigFromFile(String filePath) throws Invali
 
         try {
             JsonObject content = retriever.getConfig().toCompletionStage().toCompletableFuture().join();
-            return content.mapTo(AppConfiguration.class);
+            AppConfiguration configuration = content.mapTo(AppConfiguration.class);
+            configuration.validate();
+            return configuration;
         } catch (Exception e) {
             throw new InvalidConfigException("Failed to load Application Configuration", e);
         } finally {

server/src/main/java/com/flipkart/varadhi/config/AppConfiguration.java

@@ -1,6 +1,7 @@
 package com.flipkart.varadhi.config;
 
 import com.flipkart.varadhi.auth.AuthenticationOptions;
+import com.flipkart.varadhi.entities.Validatable;
 import com.flipkart.varadhi.spi.authz.AuthorizationOptions;
 import com.flipkart.varadhi.controller.config.ControllerConfig;
 import com.flipkart.varadhi.produce.config.ProducerOptions;
@@ -9,13 +10,14 @@
 import io.vertx.core.DeploymentOptions;
 import io.vertx.core.VertxOptions;
 import io.vertx.core.http.HttpServerOptions;
+import jakarta.validation.Valid;
 import jakarta.validation.constraints.NotNull;
 import lombok.Getter;
 
 import java.util.Map;
 
 @Getter
-public class AppConfiguration {
+public class AppConfiguration implements Validatable {
     @NotNull
     private VertxOptions vertxOptions;
 
@@ -70,4 +72,14 @@ public class AppConfiguration {
     private ControllerConfig controller;
 
     private Map<String, String> otelOptions;
+
+    @NotNull
+    @Valid
+    private MessageHeaderConfiguration messageHeaderConfiguration;
+
+    @Override
+    public void validate(){
+        Validatable.super.validate();
+        messageHeaderConfiguration.validate();
+    }
 }

server/src/main/java/com/flipkart/varadhi/config/MessageHeaderConfiguration.java

@@ -0,0 +1,98 @@
+package com.flipkart.varadhi.config;
+
+import com.flipkart.varadhi.entities.Validatable;
+import jakarta.validation.constraints.NotNull;
+import lombok.*;
+
+import java.lang.reflect.Method;
+import java.util.List;
+
+
+@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+public class MessageHeaderConfiguration implements Validatable {
+    @NotNull
+    private List<String> allowedPrefix;
+
+    // Callback codes header
+    @NotNull
+    private String callbackCodes;
+
+    // Consumer timeout header (indefinite message consumer timeout)
+    @NotNull
+    private String requestTimeout;
+
+    // HTTP related headers
+    @NotNull
+    private String replyToHttpUriHeader;
+
+    @NotNull
+    private String replyToHttpMethodHeader;
+
+    @NotNull
+    private String replyToHeader;
+
+    @NotNull
+    private String httpUriHeader;
+
+    @NotNull
+    private String httpMethodHeader;
+
+    @NotNull
+    private String httpContentType;
+
+    // Group ID & Msg ID header used to correlate messages
+    @NotNull
+    private String groupIdHeader;
+
+    @NotNull
+    private String msgIdHeader;
+
+    /**
+     * We use reflection to dynamically invoke getter methods for all fields in the
+     * `MessageHeaderConfiguration` class, allowing us to validate them without
+     * explicitly coding checks for each field. This makes the validation process
+     * scalable and easier to maintain, as any new fields with getters will be
+     * automatically validated. The `allowedPrefix` field is skipped as it is handled separately.
+     */
+    @SneakyThrows
+    @Override
+    public void validate() {
+        List<String> allowedPrefixList = getAllowedPrefix();
+        for (String prefix : allowedPrefixList) {
+            if (prefix.isEmpty() || prefix.isBlank()) {
+                throw new IllegalArgumentException("Header prefix cannot be blank");
+            }
+        }
+
+        for (Method method : MessageHeaderConfiguration.class.getDeclaredMethods()) {
+            if (isGetterMethod(method)) {
+                    if ("getAllowedPrefix".equals(method.getName())) {
+                        continue;
+                    }
+                    Object value = method.invoke(this);
+                    if (value instanceof String stringValue) {
+                        if (!startsWithValidPrefix(allowedPrefixList, stringValue)) {
+                            throw new IllegalArgumentException(method.getName() + " does not have a valid header value : " + stringValue);
+                        }
+                    }
+            }
+        }
+    }
+
+    private boolean isGetterMethod(Method method) {
+        return method.getName().startsWith("get") && method.getReturnType() != void.class;
+    }
+
+    private boolean startsWithValidPrefix(List<String> allowedPrefixes, String value) {
+        for (String prefix : allowedPrefixes) {
+            if (value.startsWith(prefix)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+}

server/src/test/java/com/flipkart/varadhi/config/MessageHeaderConfigurationTest.java

@@ -0,0 +1,52 @@
+package com.flipkart.varadhi.config;
+
+import org.junit.jupiter.api.function.Executable;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import java.util.Arrays;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+public class MessageHeaderConfigurationTest {
+
+    private MessageHeaderConfiguration getDefaultMessageHeaderConfig() {
+        return MessageHeaderConfiguration.builder()
+                .allowedPrefix(Arrays.asList("VARADHI_", "VARADHI-"))
+                .callbackCodes("VARADHI_CALLBACK_CODES")
+                .requestTimeout("VARADHI_REQUEST_TIMEOUT")
+                .replyToHttpUriHeader("VARADHI_REPLY_TO_HTTP_URI")
+                .replyToHttpMethodHeader("VARADHI_REPLY_TO_HTTP_METHOD")
+                .replyToHeader("VARADHI_REPLY_TO")
+                .httpUriHeader("VARADHI_HTTP_URI")
+                .httpMethodHeader("VARADHI_HTTP_METHOD")
+                .httpContentType("VARADHI_CONTENT_TYPE")
+                .groupIdHeader("VARADHI_GROUP_ID")
+                .msgIdHeader("VARADHI_MSG_ID")
+                .build();
+    }
+
+    @ParameterizedTest
+    @CsvSource({
+            "'VARADHI_', 'VARADHI-', true",            // Valid case
+            "'', 'VARADHI-', false",                   // Empty prefix
+            "'VARADHI_', '', false",                   // Empty second prefix
+            "'T_', 'T-', false",                       // Invalid prefix not matching
+            "'VARADHI_', null, true",                  // Null second prefix, valid first
+            "'varadhi_', 'VARADHI-', false",           // Case sensitivity issue
+            "'VARADHI_', 'VARADHI\u00A9', true",       // Unicode characters
+    })
+    void testHeaderPrefixValidation(String prefix1, String prefix2, boolean expectedResult) {
+        MessageHeaderConfiguration config = getDefaultMessageHeaderConfig();
+        config.setAllowedPrefix(Arrays.asList(prefix1, prefix2));
+
+        Executable validationAction = config::validate;
+
+        if (expectedResult) {
+            assertDoesNotThrow(validationAction, "Expected validation to pass but it failed.");
+        } else {
+            // If expected result is false, it should throw an IllegalArgumentException
+            assertThrows(IllegalArgumentException.class, validationAction, "Expected validation to throw an exception.");
+        }
+    }
+}