Added functionality to allow user to add a message to a topic partition manually

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

@@ -29,13 +29,18 @@
 import javax.validation.constraints.NotNull;
 
 import kafdrop.util.*;
+
+import org.apache.kafka.clients.producer.RecordMetadata;
 import org.springframework.http.MediaType;
 import org.springframework.stereotype.Controller;
 import org.springframework.ui.Model;
 import org.springframework.validation.BindingResult;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.ModelAttribute;
 import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.PutMapping;
+import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RequestMethod;
 import org.springframework.web.bind.annotation.RequestParam;
@@ -53,6 +58,7 @@
 import kafdrop.config.ProtobufDescriptorConfiguration.ProtobufDescriptorProperties;
 import kafdrop.config.SchemaRegistryConfiguration;
 import kafdrop.config.SchemaRegistryConfiguration.SchemaRegistryProperties;
+import kafdrop.model.CreateMessageVO;
 import kafdrop.model.MessageVO;
 import kafdrop.model.TopicPartitionVO;
 import kafdrop.model.TopicVO;
@@ -179,6 +185,54 @@ public String viewMessageForm(@PathVariable("name") String topicName,
 
     return "message-inspector";
   }
+
+  @PostMapping("/topic/{name:.+}/addmessage")
+  public String addMessage(@PathVariable("name") String topicName, @ModelAttribute("addMessageForm") CreateMessageVO body, Model model) {
+	  try {
+		  final MessageFormat defaultFormat = messageFormatProperties.getFormat();
+		  final MessageFormat defaultKeyFormat = keyFormatProperties.getFormat();
+
+		  final var serializers = new Serializers(
+		          getSerializer(topicName, defaultKeyFormat, "", ""),
+		          getSerializer(topicName, defaultFormat, "", ""));
+		  RecordMetadata recordMetadata = kafkaMonitor.publishMessage(body, serializers);
+
+		  final var deserializers = new Deserializers(
+		          getDeserializer(topicName, defaultKeyFormat, "", ""),
+		          getDeserializer(topicName, defaultFormat, "", "")
+		      );
+
+		  final PartitionOffsetInfo defaultForm = new PartitionOffsetInfo();
+
+	      defaultForm.setCount(100l);
+	      defaultForm.setOffset(recordMetadata.offset());
+	      defaultForm.setPartition(body.getTopicPartition());
+	      defaultForm.setFormat(defaultFormat);
+	      defaultForm.setKeyFormat(defaultFormat);
+
+	      model.addAttribute("messageForm", defaultForm);
+
+		  final TopicVO topic = kafkaMonitor.getTopic(topicName)
+			        .orElseThrow(() -> new TopicNotFoundException(topicName));
+
+		  model.addAttribute("topic", topic);
+
+		  model.addAttribute("defaultFormat", defaultFormat);
+		  model.addAttribute("messageFormats", MessageFormat.values());
+		  model.addAttribute("defaultKeyFormat", defaultKeyFormat);
+		  model.addAttribute("keyFormats",KeyFormat.values());
+		  model.addAttribute("descFiles", protobufProperties.getDescFilesList());
+		  model.addAttribute("messages",
+                  messageInspector.getMessages(topicName,
+                                               body.getTopicPartition(),
+                                               recordMetadata.offset(),
+                                               100,
+                                               deserializers));
+	  } catch(Exception ex) {
+		  model.addAttribute("errorMessage", ex.getMessage());
+	  }
+	  return "message-inspector";
+  }
 
 
   /**
@@ -278,6 +332,30 @@ private MessageDeserializer getDeserializer(String topicName, MessageFormat form
 
     return deserializer;
   }
+
+  private MessageSerializer getSerializer(String topicName, MessageFormat format, String descFile, String msgTypeName) {
+	  final MessageSerializer serializer;
+
+	  if (format == MessageFormat.AVRO) {
+	      final var schemaRegistryUrl = schemaRegistryProperties.getConnect();
+	      final var schemaRegistryAuth = schemaRegistryProperties.getAuth();
+
+	      serializer = new AvroMessageSerializer(topicName, schemaRegistryUrl, schemaRegistryAuth);
+	    } else if (format == MessageFormat.PROTOBUF) {
+	      // filter the input file name
+	      final var descFileName = descFile.replace(".desc", "")
+	          .replaceAll("\\.", "")
+	          .replaceAll("/", "");
+	      final var fullDescFile = protobufProperties.getDirectory() + File.separator + descFileName + ".desc";
+	      serializer = new ProtobufMessageSerializer(fullDescFile, msgTypeName);
+	    } else if (format == MessageFormat.MSGPACK) {
+	    	serializer = new MsgPackMessageSerializer();
+	    } else {
+	    	serializer = new DefaultMessageSerializer();
+	    }
+
+	  return serializer;
+  }
 
   /**
    * Encapsulates offset data for a single partition.

src/main/java/kafdrop/model/CreateMessageVO.java

@@ -0,0 +1,24 @@
+package kafdrop.model;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiParam;
+import lombok.Data;
+import lombok.RequiredArgsConstructor;
+
+@Data
+@RequiredArgsConstructor
+@ApiModel("Create message model")
+public final class CreateMessageVO {
+
+	@ApiParam("Topic partition")
+	private int topicPartition;
+
+	@ApiParam("Topic key")
+	private String key;
+
+	@ApiParam("Topic value")
+	private String value;
+
+	@ApiParam("Topic name")
+	private String topic;
+}

src/main/java/kafdrop/service/KafkaHighLevelProducer.java

@@ -0,0 +1,78 @@
+package kafdrop.service;
+
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
+
+import javax.annotation.PostConstruct;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.ByteArrayDeserializer;
+import org.apache.kafka.common.serialization.ByteArraySerializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+
+import kafdrop.config.KafkaConfiguration;
+import kafdrop.model.CreateMessageVO;
+import kafdrop.util.Serializers;
+
+@Service
+public final class KafkaHighLevelProducer {
+
+	private static final Logger LOG = LoggerFactory.getLogger(KafkaHighLevelProducer.class);
+
+	private KafkaProducer<byte[], byte[]> kafkaProducer;
+
+	private final KafkaConfiguration kafkaConfiguration;
+
+	public KafkaHighLevelProducer(KafkaConfiguration kafkaConfiguration) {
+		this.kafkaConfiguration = kafkaConfiguration;
+	}
+
+	@PostConstruct
+	private void initializeClient() {
+		if (kafkaProducer == null) {
+			final var properties = new Properties();
+			properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
+			properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);
+			properties.put(ProducerConfig.ACKS_CONFIG, "all");
+			properties.put(ProducerConfig.RETRIES_CONFIG, 0);
+			properties.put(ProducerConfig.LINGER_MS_CONFIG, 1);
+			properties.put(ProducerConfig.CLIENT_ID_CONFIG, "kafdrop-producer");
+			kafkaConfiguration.applyCommon(properties);
+
+			kafkaProducer = new KafkaProducer<>(properties);
+		}
+	}
+
+	public RecordMetadata publishMessage(CreateMessageVO message, Serializers searilazers) {
+		initializeClient();
+
+		final ProducerRecord<byte[], byte[]> record = new ProducerRecord<byte[], byte[]>(message.getTopic(),
+				message.getTopicPartition(),
+				searilazers.getKeySerializer().serializeMessage(message.getKey()),
+				searilazers.getValueSerializer().serializeMessage(message.getValue()));
+
+		Future<RecordMetadata> result = kafkaProducer.send(record);
+		try {
+			RecordMetadata recordMetadata = result.get();
+			LOG.info("Record published successfully [{}]", recordMetadata);
+			return recordMetadata;
+		} catch (Exception e) {
+			LOG.error("Failed to publish message", e);
+			throw new KafkaProducerException(e);
+		}
+	}
+}

src/main/java/kafdrop/service/KafkaMonitor.java

@@ -20,6 +20,8 @@
 
 import kafdrop.model.*;
 import kafdrop.util.*;
+
+import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.*;
 
 import java.util.*;
@@ -57,6 +59,8 @@ List<MessageVO> getMessages(TopicPartition topicPartition, long offset, int coun
    * @param topic name of the topic to delete
    */
   void deleteTopic(String topic);
+
+  RecordMetadata publishMessage(CreateMessageVO message, Serializers serializers);
 
   List<AclVO> getAcls();
 }

src/main/java/kafdrop/service/KafkaMonitorImpl.java

@@ -23,6 +23,7 @@
 import org.apache.kafka.clients.admin.*;
 import org.apache.kafka.clients.admin.ConfigEntry.*;
 import org.apache.kafka.clients.consumer.*;
+import org.apache.kafka.clients.producer.RecordMetadata;
 import org.apache.kafka.common.*;
 import org.apache.kafka.common.header.*;
 import org.slf4j.*;
@@ -42,10 +43,13 @@ public final class KafkaMonitorImpl implements KafkaMonitor {
   private final KafkaHighLevelConsumer highLevelConsumer;
 
   private final KafkaHighLevelAdminClient highLevelAdminClient;
+
+  private final KafkaHighLevelProducer highLevelProducer;
 
-  public KafkaMonitorImpl(KafkaHighLevelConsumer highLevelConsumer, KafkaHighLevelAdminClient highLevelAdminClient) {
+  public KafkaMonitorImpl(KafkaHighLevelConsumer highLevelConsumer, KafkaHighLevelAdminClient highLevelAdminClient, KafkaHighLevelProducer highLevelProducer) {
     this.highLevelConsumer = highLevelConsumer;
     this.highLevelAdminClient = highLevelAdminClient;
+    this.highLevelProducer = highLevelProducer;
   }
 
   @Override
@@ -315,4 +319,9 @@ private List<ConsumerGroupOffsets> getConsumerOffsets(Set<String> topics) {
         .filter(not(ConsumerGroupOffsets::isEmpty))
         .collect(Collectors.toList());
   }
+
+	@Override
+	public RecordMetadata publishMessage(CreateMessageVO message, Serializers serializers) {
+		return highLevelProducer.publishMessage(message, serializers);
+	}
 }

src/main/java/kafdrop/service/KafkaProducerException.java

@@ -0,0 +1,16 @@
+package kafdrop.service;
+
+import org.springframework.http.HttpStatus;
+import org.springframework.web.bind.annotation.ResponseStatus;
+
+@ResponseStatus(HttpStatus.INTERNAL_SERVER_ERROR)
+public class KafkaProducerException extends RuntimeException {
+
+	public KafkaProducerException(Throwable exception) {
+		super(exception);
+	}
+
+	public KafkaProducerException(String message) {
+		super(message);
+	}
+}

src/main/java/kafdrop/util/AvroMessageSerializer.java

@@ -0,0 +1,36 @@
+package kafdrop.util;
+
+import java.util.HashMap;
+
+import io.confluent.kafka.serializers.AbstractKafkaAvroSerDeConfig;
+import io.confluent.kafka.serializers.KafkaAvroSerializer;
+
+public class AvroMessageSerializer implements MessageSerializer {
+
+	private final String topicName;
+	private final KafkaAvroSerializer serializer;
+
+	public AvroMessageSerializer(String topicName, String schemaRegistryUrl, String schemaRegistryAuth) {
+		this.topicName = topicName;
+		this.serializer = getSerializer(schemaRegistryUrl, schemaRegistryAuth);
+	}
+
+	@Override
+	public byte[] serializeMessage(String value) {
+		final var bytes = value.getBytes();
+		return serializer.serialize(topicName, bytes);
+	}
+
+	private KafkaAvroSerializer getSerializer(String schemaRegistryUrl, String schemaRegistryAuth) {
+		final var config = new HashMap<String, Object>();
+	    config.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
+	    if (schemaRegistryAuth != null) {
+	      config.put(AbstractKafkaAvroSerDeConfig.BASIC_AUTH_CREDENTIALS_SOURCE, "USER_INFO");
+	      config.put(AbstractKafkaAvroSerDeConfig.USER_INFO_CONFIG, schemaRegistryAuth);
+	    }
+	    final var kafkaAvroSerializer = new KafkaAvroSerializer();
+	    kafkaAvroSerializer.configure(config, false);
+	    return kafkaAvroSerializer;
+	}
+
+}

src/main/java/kafdrop/util/DefaultMessageSerializer.java

@@ -0,0 +1,10 @@
+package kafdrop.util;
+
+public class DefaultMessageSerializer implements MessageSerializer{
+
+	@Override
+	public byte[] serializeMessage(String value) {
+		return value.getBytes();
+	}
+
+}

src/main/java/kafdrop/util/MessageSerializer.java

@@ -0,0 +1,6 @@
+package kafdrop.util;
+
+@FunctionalInterface
+public interface MessageSerializer {
+	byte[] serializeMessage(String value);
+}

src/main/java/kafdrop/util/MsgPackMessageSerializer.java

@@ -0,0 +1,26 @@
+package kafdrop.util;
+
+import java.io.IOException;
+
+import org.msgpack.core.MessageBufferPacker;
+import org.msgpack.core.MessagePack;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MsgPackMessageSerializer implements MessageSerializer{
+
+	private static final Logger LOGGER = LoggerFactory.getLogger(MsgPackMessageSerializer.class);
+
+	@Override
+	public byte[] serializeMessage(String value) {
+		try (MessageBufferPacker packer = MessagePack.newDefaultBufferPacker()){
+			packer.packString(value);
+			return packer.toByteArray();
+		} catch (IOException e) {
+			final String errorMsg = "Unable to pack msgpack message";
+		    LOGGER.error(errorMsg, e);
+		    throw new DeserializationException(errorMsg);
+		}
+	}
+
+}

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

@@ -64,7 +64,7 @@ public String deserializeMessage(ByteBuffer buffer) {
         throw new DeserializationException(errorMsg);
       }
       DynamicMessage message = DynamicMessage.parseFrom(messageDescriptor.get(), CodedInputStream.newInstance(buffer));
-
+      
       JsonFormat.TypeRegistry typeRegistry = JsonFormat.TypeRegistry.newBuilder().add(descriptors).build();
       Printer printer = JsonFormat.printer().usingTypeRegistry(typeRegistry);