[allow_configuring_avro_consumer] Debugging / fixes

Author: Yevgeniy Magdel

src/main/java/kafdrop/config/MessageFormatConfiguration.java

@@ -31,4 +31,26 @@ public void setFormat(MessageFormat format) {
       this.format = format;
     }
   }
+
+  @Component
+  @ConfigurationProperties(prefix = "key")
+  public static final class KeyFormatProperties {
+    private MessageFormat format;
+
+    @PostConstruct
+    public void init() {
+      // Set a default message format if not configured.
+      if (format == null) {
+        format = MessageFormat.DEFAULT;
+      }
+    }
+
+    public MessageFormat getFormat() {
+      return format;
+    }
+
+    public void setFormat(MessageFormat format) {
+      this.format = format;
+    }
+  }
 }

src/main/java/kafdrop/controller/MessageController.java

@@ -28,6 +28,7 @@
 import javax.validation.constraints.Min;
 import javax.validation.constraints.NotNull;
 
+import kafdrop.config.MessageFormatConfiguration;
 import kafdrop.util.*;
 import org.springframework.http.MediaType;
 import org.springframework.stereotype.Controller;
@@ -64,19 +65,19 @@ public final class MessageController {
   private final MessageInspector messageInspector;
 
   private final MessageFormatProperties messageFormatProperties;
-  private final MessageFormatProperties keyFormatProperties;
+  private final MessageFormatConfiguration.KeyFormatProperties keyFormatProperties;
 
   private final SchemaRegistryProperties schemaRegistryProperties;
 
   private final ProtobufDescriptorProperties protobufProperties;
 
-  public MessageController(KafkaMonitor kafkaMonitor, MessageInspector messageInspector, MessageFormatProperties messageFormatProperties, MessageFormatProperties keyFormatProperties, SchemaRegistryProperties schemaRegistryProperties, ProtobufDescriptorProperties protobufProperties) {
+  public MessageController(KafkaMonitor kafkaMonitor, MessageInspector messageInspector, MessageFormatProperties messageFormatProperties, MessageFormatConfiguration.KeyFormatProperties keyFormatProperties, SchemaRegistryProperties schemaRegistryProperties, ProtobufDescriptorProperties protobufProperties) {
     this.kafkaMonitor = kafkaMonitor;
     this.messageInspector = messageInspector;
     this.messageFormatProperties = messageFormatProperties;
     this.keyFormatProperties = keyFormatProperties;
     this.schemaRegistryProperties = schemaRegistryProperties;
-    this.protobufProperties = protobufProperties; 
+    this.protobufProperties = protobufProperties;
   }
 
   /**
@@ -314,9 +315,9 @@ public static class PartitionOffsetInfo {
     private MessageFormat format;
 
     private MessageFormat keyFormat;
-    
+
     private String descFile;
-    
+
     private String msgTypeName;
 
     public PartitionOffsetInfo(int partition, long offset, long count, MessageFormat format) {

src/main/java/kafdrop/util/AvroMessageDeserializer.java

@@ -37,18 +37,18 @@ private static KafkaAvroDeserializer getDeserializer(String schemaRegistryUrl, S
   }
 
   private static void setConfigFromEnvIfAvailable(String topicName, String configPath, Map<String,Object> config){
-
     String configPrefix = "SCHEMA_REGISTRY";
-    String topicScopedEnvPath = Arrays.stream(new String[]{configPrefix, configPath.replace(".", "_"), topicName.replace("-", "_") } )
+    String topicScopedEnvPath = Arrays.stream(new String[]{configPrefix, configPath.replaceAll("\\.", "_"), topicName.replaceAll("-", "_") } )
             .map(String::toUpperCase).collect(Collectors.joining("_"));
 
-    String noTopicScopedEnvPath = Arrays.stream(new String[]{ "SCHEMA_REGISTRY", configPath.replace(".", "_") })
+    String noTopicScopedEnvPath = Arrays.stream(new String[]{ configPrefix, configPath.replaceAll("\\.", "_") })
             .map(String::toUpperCase).collect(Collectors.joining("_"));
 
     for(String envPath : new String[]{topicScopedEnvPath, noTopicScopedEnvPath}) {
+
       String namingStrategyValue = System.getenv(envPath);
       if (namingStrategyValue != null) {
-        config.put(envPath, namingStrategyValue);
+        config.put(configPath, namingStrategyValue);
       }
     }
   }