Support Protobuf "Any" message type for unmixed topics

src/main/java/kafdrop/config/ProtobufDescriptorConfiguration.java

@@ -26,6 +26,8 @@ public static final class ProtobufDescriptorProperties {
     // detail screen
     private String directory;
 
+    private Boolean parseAnyProto = Boolean.FALSE;
+
     public String getDirectory() {
       return directory;
     }
@@ -34,6 +36,14 @@ public void setDirectory(String directory) {
       this.directory = directory;
     }
 
+    public Boolean getParseAnyProto() {
+      return parseAnyProto;
+    }
+
+    public void setParseAnyProto(Boolean parseAnyProto) {
+      this.parseAnyProto = parseAnyProto;
+    }
+
     public List<String> getDescFilesList() {
       // getting file list
       if (directory == null || Files.notExists(Path.of(directory))) {

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

@@ -99,8 +99,8 @@ public String viewAllMessages(@PathVariable("name") String topicName,
     model.addAttribute("descFiles", protobufProperties.getDescFilesList());
 
     final var deserializers = new Deserializers(
-          getDeserializer(topicName, defaultKeyFormat, "", ""),
-          getDeserializer(topicName, defaultFormat, "", ""));
+          getDeserializer(topicName, defaultKeyFormat, "", "", protobufProperties.getParseAnyProto()),
+          getDeserializer(topicName, defaultFormat, "", "", protobufProperties.getParseAnyProto()));
 
     final List<MessageVO> messages = messageInspector.getMessages(topicName, size, deserializers);
 
@@ -142,6 +142,7 @@ public String viewMessageForm(@PathVariable("name") String topicName,
       defaultForm.setPartition(0);
       defaultForm.setFormat(defaultFormat);
       defaultForm.setKeyFormat(defaultFormat);
+      defaultForm.setIsAnyProto(protobufProperties.getParseAnyProto());
 
       model.addAttribute("messageForm", defaultForm);
     }
@@ -155,12 +156,13 @@ public String viewMessageForm(@PathVariable("name") String topicName,
     model.addAttribute("defaultKeyFormat", defaultKeyFormat);
     model.addAttribute("keyFormats",KeyFormat.values());
     model.addAttribute("descFiles", protobufProperties.getDescFilesList());
+    model.addAttribute("isAnyProtoOpts", List.of(true, false));
 
     if (!messageForm.isEmpty() && !errors.hasErrors()) {
 
       final var deserializers = new Deserializers(
-          getDeserializer(topicName, messageForm.getKeyFormat(), messageForm.getDescFile(),messageForm.getMsgTypeName()),
-          getDeserializer(topicName, messageForm.getFormat(), messageForm.getDescFile(), messageForm.getMsgTypeName())
+          getDeserializer(topicName, messageForm.getKeyFormat(), messageForm.getDescFile(),messageForm.getMsgTypeName(), messageForm.getIsAnyProto()),
+          getDeserializer(topicName, messageForm.getFormat(), messageForm.getDescFile(), messageForm.getMsgTypeName(), messageForm.getIsAnyProto())
       );
 
       model.addAttribute("messages",
@@ -218,7 +220,8 @@ List<Object> getPartitionOrMessages(
       @RequestParam(name = "format", required = false) String format,
       @RequestParam(name = "keyFormat", required = false) String keyFormat,
       @RequestParam(name = "descFile", required = false) String descFile,
-      @RequestParam(name = "msgTypeName", required = false) String msgTypeName
+      @RequestParam(name = "msgTypeName", required = false) String msgTypeName,
+      @RequestParam(name = "isAnyProto", required = false) Boolean isAnyProto
   ) {
     if (partition == null || offset == null || count == null) {
       final TopicVO topic = kafkaMonitor.getTopic(topicName)
@@ -231,8 +234,8 @@ List<Object> getPartitionOrMessages(
     } else {
 
       final var deserializers = new Deserializers(
-              getDeserializer(topicName, getSelectedMessageFormat(keyFormat), descFile, msgTypeName),
-              getDeserializer(topicName, getSelectedMessageFormat(format), descFile, msgTypeName));
+              getDeserializer(topicName, getSelectedMessageFormat(keyFormat), descFile, msgTypeName, isAnyProto),
+              getDeserializer(topicName, getSelectedMessageFormat(format), descFile, msgTypeName, isAnyProto));
 
       List<Object> messages = new ArrayList<>();
       List<MessageVO> vos = messageInspector.getMessages(
@@ -250,7 +253,7 @@ List<Object> getPartitionOrMessages(
     }
   }
 
-  private MessageDeserializer getDeserializer(String topicName, MessageFormat format, String descFile, String msgTypeName) {
+  private MessageDeserializer getDeserializer(String topicName, MessageFormat format, String descFile, String msgTypeName, boolean isAnyProto) {
     final MessageDeserializer deserializer;
 
     if (format == MessageFormat.AVRO) {
@@ -265,7 +268,7 @@ private MessageDeserializer getDeserializer(String topicName, MessageFormat form
           .replace(".", "")
           .replace("/", "");
       final var fullDescFile = protobufProperties.getDirectory() + File.separator + descFileName + ".desc";
-      deserializer = new ProtobufMessageDeserializer(fullDescFile, msgTypeName);
+      deserializer = new ProtobufMessageDeserializer(fullDescFile, msgTypeName, isAnyProto);
     } else if (format == MessageFormat.PROTOBUF) {
       final var schemaRegistryUrl = schemaRegistryProperties.getConnect();
       final var schemaRegistryAuth = schemaRegistryProperties.getAuth();
@@ -317,6 +320,8 @@ public static class PartitionOffsetInfo {
 
     private String msgTypeName;
 
+    private Boolean isAnyProto = Boolean.FALSE;
+
     public PartitionOffsetInfo(int partition, long offset, long count, MessageFormat format) {
       this.partition = partition;
       this.offset = offset;
@@ -391,5 +396,13 @@ public void setMsgTypeName(String msgTypeName) {
       this.msgTypeName = msgTypeName;
     }
 
+    public Boolean getIsAnyProto() {
+      return isAnyProto;
+    }
+
+    public void setIsAnyProto(Boolean isAnyProto) {
+      this.isAnyProto = isAnyProto;
+    }
+
   }
 }

src/main/java/kafdrop/util/ProtobufMessageDeserializer.java

@@ -7,11 +7,13 @@
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.protobuf.Any;
 import com.google.protobuf.CodedInputStream;
 import com.google.protobuf.DescriptorProtos.FileDescriptorProto;
 import com.google.protobuf.DescriptorProtos.FileDescriptorSet;
@@ -26,17 +28,20 @@ public class ProtobufMessageDeserializer implements MessageDeserializer {
 
   private final String fullDescFile;
   private final String msgTypeName;
+  private final boolean isAnyProto;
 
   private static final Logger LOG = LoggerFactory.getLogger(ProtobufMessageDeserializer.class);
 
-  public ProtobufMessageDeserializer(String fullDescFile, String msgTypeName) {
+  public ProtobufMessageDeserializer(String fullDescFile, String msgTypeName, boolean isAnyProto) {
     this.fullDescFile = fullDescFile;
     this.msgTypeName = msgTypeName;
+    this.isAnyProto = isAnyProto;
   }
 
   @Override
   public String deserializeMessage(ByteBuffer buffer) {
 
+    AtomicReference<String> msgTypeNameRef = new AtomicReference<>(msgTypeName);
     try (InputStream input = new FileInputStream(fullDescFile)) {
       FileDescriptorSet set = FileDescriptorSet.parseFrom(input);
 
@@ -49,14 +54,27 @@ public String deserializeMessage(ByteBuffer buffer) {
       }
 
       final var descriptors = descs.stream().flatMap(desc -> desc.getMessageTypes().stream()).collect(Collectors.toList());
-
-      final var messageDescriptor = descriptors.stream().filter(desc -> msgTypeName.equals(desc.getName())).findFirst();
+      // automatically detect the message type name if the proto is "Any" and no message type name is given
+      if (isAnyProto && msgTypeName.isBlank()) {
+        String typeUrl = Any.parseFrom(buffer).getTypeUrl();
+				String[] splittedTypeUrl = typeUrl.split("/");
+				// the last part in the type url is always the FQCN for this proto
+				msgTypeNameRef.set(splittedTypeUrl[splittedTypeUrl.length - 1]);
+      }
+      // check for full name too if the proto is "Any"
+      final var messageDescriptor = descriptors.stream().filter(desc -> msgTypeNameRef.get().equals(desc.getName()) || msgTypeNameRef.get().equals(desc.getFullName())).findFirst();
       if (messageDescriptor.isEmpty()) {
-        final String errorMsg = "Can't find specific message type: " + msgTypeName;
+        final String errorMsg = "Can't find specific message type: " + msgTypeNameRef.get();
         LOG.error(errorMsg);
         throw new DeserializationException(errorMsg);
       }
-      DynamicMessage message = DynamicMessage.parseFrom(messageDescriptor.get(), CodedInputStream.newInstance(buffer));
+      DynamicMessage message = null;
+      if (isAnyProto) {
+        // parse the value from "Any" proto instead of the "Any" proto itself
+        message = DynamicMessage.parseFrom(messageDescriptor.get(), Any.parseFrom(buffer).getValue());
+      } else {
+        message = DynamicMessage.parseFrom(messageDescriptor.get(), CodedInputStream.newInstance(buffer));
+      }
 
       JsonFormat.TypeRegistry typeRegistry = JsonFormat.TypeRegistry.newBuilder().add(descriptors).build();
       Printer printer = JsonFormat.printer().usingTypeRegistry(typeRegistry);
@@ -71,7 +89,7 @@ public String deserializeMessage(ByteBuffer buffer) {
       LOG.error(errorMsg, e);
       throw new DeserializationException(errorMsg);
     } catch (DescriptorValidationException e) {
-      final String errorMsg = "Can't compile proto message type: " + msgTypeName;
+      final String errorMsg = "Can't compile proto message type: " + msgTypeNameRef.get();
       LOG.error(errorMsg, e);
       throw new DeserializationException(errorMsg);
     }

src/main/resources/templates/message-inspector.ftlh

@@ -49,12 +49,14 @@
 	    {
 	        $('#protobufDescriptor').hide();
 	        $('#protobufMsgType').hide();
+            $('#protobufIsAnyProto').hide();
 	    }
 	    else
 	    {
 	        $('#protobufDescriptor').show();
 	        $('#protobufMsgType').show();
 	        $('#descFile').prop('required',true);
+            $('#protobufIsAnyProto').show();
 	    }
 	}  
 	$(document).ready(function(){
@@ -72,6 +74,7 @@
 <#assign selectedPartition=messageForm.partition!0?number>
 <#assign selectedFormat=messageForm.format!defaultFormat>
 <#assign selectedKeyFormat=messageForm.keyFormat!defaultFormat>
+<#assign selectedIsAnyProtoOpt=messageForm.isAnyProto>
 
 <div id="partitionSizes">
     <#assign curPartition=topic.getPartition(selectedPartition).get()>
@@ -149,6 +152,15 @@
             </#if>
         </div>
         &nbsp;&nbsp;
+        <div class="form-group" id="protobufIsAnyProto">
+        	<label class=control-label" for="format">Is Any proto?</label>
+            <select class="form-control" id="isAnyProto" name="isAnyProto">
+                 <#list isAnyProtoOpts as opt>
+                    <option value="${opt?c}" <#if opt == selectedIsAnyProtoOpt>selected="selected"</#if>>${opt?string('Yes', 'No')}</option>
+                </#list>
+            </select>
+        </div>
+        &nbsp;&nbsp;
         <button id="viewMessagesBtn" class="btn btn-success" type="submit" ><i class="fa fa-search"></i> View Messages</button>
     </form>
 </div>

src/test/java/kafdrop/AbstractIntegrationTest.java

@@ -20,7 +20,7 @@ static class Initializer implements ApplicationContextInitializer<ConfigurableAp
 
         public static Map<String, Object> getProperties() {
             Startables.deepStart(List.of(kafka)).join();
-            return Map.of("kafka.brokerConnect", kafka.getBootstrapServers());
+			return Map.of("kafka.brokerConnect", kafka.getBootstrapServers(), "protobufdesc.directory","./src/test/resources", "protobufdesc.parseAnyProto", true);
         }
 
         @Override

src/test/java/kafdrop/SendKafkaProtoPayload.java

@@ -0,0 +1,39 @@
+package kafdrop;
+
+import java.util.List;
+import java.util.UUID;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.context.event.ApplicationReadyEvent;
+import org.springframework.boot.test.context.TestComponent;
+import org.springframework.context.ApplicationListener;
+import org.springframework.kafka.core.KafkaTemplate;
+
+import com.google.protobuf.Any;
+
+import kafdrop.kafka.KafkaTestConfiguration;
+import kafdrop.protos.Person;
+
+@TestComponent
+public class SendKafkaProtoPayload implements ApplicationListener<ApplicationReadyEvent> {
+
+	@Autowired
+	private KafkaTemplate<String, byte[]> kafkaTemplate;
+
+	@Override
+	public void onApplicationEvent(ApplicationReadyEvent event) {
+		if (!AbstractIntegrationTest.Initializer.kafka.isRunning()) {
+			throw new IllegalStateException("Kafka container not started");
+		}
+
+		Person person = Person.newBuilder()
+				.setName("Max Mustermann")
+				.setId(1)
+				.setEmail("[email protected]")
+				.setContact(Person.Contact.MOBILE)
+				.addAllData(List.of("This", "is", "a", "test"))
+				.build();
+		byte[] payload = Any.pack(person).toByteArray();
+		kafkaTemplate.send(KafkaTestConfiguration.TEST_TOPIC_NAME, UUID.randomUUID().toString(), payload);
+	}
+}

src/test/java/kafdrop/kafka/KafkaTestConfiguration.java

@@ -0,0 +1,53 @@
+package kafdrop.kafka;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.kafka.clients.admin.AdminClientConfig;
+import org.apache.kafka.clients.admin.NewTopic;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.test.context.TestConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.kafka.config.TopicBuilder;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaAdmin;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.core.ProducerFactory;
+
+@TestConfiguration
+public class KafkaTestConfiguration {
+
+	@Value("${kafka.brokerConnect}")
+	private String brokerConnect;
+
+	public static final String TEST_TOPIC_NAME = "kafdrop-test";
+
+	@Bean
+	public ProducerFactory<String, byte[]> producerFactory() {
+		Map<String, Object> config = new HashMap<>();
+		config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerConnect);
+		config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+		config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
+		return new DefaultKafkaProducerFactory<>(config);
+	}
+
+	@Bean
+	public KafkaTemplate<String, byte[]> kafkaTemplate() {
+		return new KafkaTemplate<>(producerFactory());
+	}
+
+	@Bean
+	public KafkaAdmin kafkaAdmin() {
+		Map<String, Object> config = new HashMap<>();
+		config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, brokerConnect);
+		return new KafkaAdmin(config);
+	}
+
+	@Bean
+	public NewTopic topic() {
+		return TopicBuilder.name(TEST_TOPIC_NAME).build();
+	}
+}