From ad8aa0f12530b44be02c1264393283b48eb2f5ef Mon Sep 17 00:00:00 2001 From: Yevgeniy Magdel Date: Wed, 17 Nov 2021 12:53:57 -0600 Subject: [PATCH 1/2] [allow_configuring_avro_consumer] Read avro consumer configs from env var --- .../kafdrop/util/AvroMessageDeserializer.java | 23 +++++++++++++++++-- 1 file changed, 21 insertions(+), 2 deletions(-) diff --git a/src/main/java/kafdrop/util/AvroMessageDeserializer.java b/src/main/java/kafdrop/util/AvroMessageDeserializer.java index b01ea072..c42ee671 100644 --- a/src/main/java/kafdrop/util/AvroMessageDeserializer.java +++ b/src/main/java/kafdrop/util/AvroMessageDeserializer.java @@ -4,6 +4,7 @@ import java.nio.*; import java.util.*; +import java.util.stream.Collectors; public final class AvroMessageDeserializer implements MessageDeserializer { @@ -12,7 +13,7 @@ public final class AvroMessageDeserializer implements MessageDeserializer { public AvroMessageDeserializer(String topicName, String schemaRegistryUrl, String schemaRegistryAuth) { this.topicName = topicName; - this.deserializer = getDeserializer(schemaRegistryUrl, schemaRegistryAuth); + this.deserializer = getDeserializer(schemaRegistryUrl, schemaRegistryAuth, topicName); } @Override @@ -22,15 +23,33 @@ public String deserializeMessage(ByteBuffer buffer) { return deserializer.deserialize(topicName, bytes).toString(); } - private static KafkaAvroDeserializer getDeserializer(String schemaRegistryUrl, String schemaRegistryAuth) { + private static KafkaAvroDeserializer getDeserializer(String schemaRegistryUrl, String schemaRegistryAuth, String topicName) { final var config = new HashMap(); config.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl); if (schemaRegistryAuth != null) { config.put(AbstractKafkaSchemaSerDeConfig.BASIC_AUTH_CREDENTIALS_SOURCE, "USER_INFO"); config.put(AbstractKafkaSchemaSerDeConfig.USER_INFO_CONFIG, schemaRegistryAuth); } + setConfigFromEnvIfAvailable(topicName, AbstractKafkaAvroSerDeConfig.VALUE_SUBJECT_NAME_STRATEGY, config); final var kafkaAvroDeserializer = new KafkaAvroDeserializer(); kafkaAvroDeserializer.configure(config, false); return kafkaAvroDeserializer; } + + private static void setConfigFromEnvIfAvailable(String topicName, String configPath, Map config){ + + String configPrefix = "SCHEMA_REGISTRY"; + String topicScopedEnvPath = Arrays.stream(new String[]{configPrefix, configPath.replace(".", "_"), topicName.replace("-", "_") } ) + .map(String::toUpperCase).collect(Collectors.joining("_")); + + String noTopicScopedEnvPath = Arrays.stream(new String[]{ "SCHEMA_REGISTRY", configPath.replace(".", "_") }) + .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); + } + } + } } From 86cc3d5565c9c23c63ee7fcfe821f44818f2d2e0 Mon Sep 17 00:00:00 2001 From: Yevgeniy Magdel Date: Wed, 17 Nov 2021 13:08:04 -0600 Subject: [PATCH 2/2] [allow_configuring_avro_consumer] Debugging / fixes --- .../config/MessageFormatConfiguration.java | 22 +++++++++++++++++++ .../kafdrop/controller/MessageController.java | 11 +++++----- .../kafdrop/util/AvroMessageDeserializer.java | 8 +++---- 3 files changed, 32 insertions(+), 9 deletions(-) diff --git a/src/main/java/kafdrop/config/MessageFormatConfiguration.java b/src/main/java/kafdrop/config/MessageFormatConfiguration.java index e43e90cf..290a7653 100644 --- a/src/main/java/kafdrop/config/MessageFormatConfiguration.java +++ b/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; + } + } } diff --git a/src/main/java/kafdrop/controller/MessageController.java b/src/main/java/kafdrop/controller/MessageController.java index 949f47e0..deccb4ff 100644 --- a/src/main/java/kafdrop/controller/MessageController.java +++ b/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; @@ -62,19 +63,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; } /** @@ -312,9 +313,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) { diff --git a/src/main/java/kafdrop/util/AvroMessageDeserializer.java b/src/main/java/kafdrop/util/AvroMessageDeserializer.java index c42ee671..b66b8c18 100644 --- a/src/main/java/kafdrop/util/AvroMessageDeserializer.java +++ b/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 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); } } }