diff --git a/src/main/java/kafdrop/controller/MessageController.java b/src/main/java/kafdrop/controller/MessageController.java index 8350e63a..7bcd85d7 100644 --- a/src/main/java/kafdrop/controller/MessageController.java +++ b/src/main/java/kafdrop/controller/MessageController.java @@ -29,6 +29,8 @@ 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; @@ -36,6 +38,9 @@ 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. diff --git a/src/main/java/kafdrop/model/CreateMessageVO.java b/src/main/java/kafdrop/model/CreateMessageVO.java new file mode 100644 index 00000000..26986830 --- /dev/null +++ b/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; +} diff --git a/src/main/java/kafdrop/service/KafkaHighLevelProducer.java b/src/main/java/kafdrop/service/KafkaHighLevelProducer.java new file mode 100644 index 00000000..dd683b1e --- /dev/null +++ b/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 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 record = new ProducerRecord(message.getTopic(), + message.getTopicPartition(), + searilazers.getKeySerializer().serializeMessage(message.getKey()), + searilazers.getValueSerializer().serializeMessage(message.getValue())); + + Future 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); + } + } +} \ No newline at end of file diff --git a/src/main/java/kafdrop/service/KafkaMonitor.java b/src/main/java/kafdrop/service/KafkaMonitor.java index d4bcc000..4d2535d8 100644 --- a/src/main/java/kafdrop/service/KafkaMonitor.java +++ b/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 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 getAcls(); } diff --git a/src/main/java/kafdrop/service/KafkaMonitorImpl.java b/src/main/java/kafdrop/service/KafkaMonitorImpl.java index ea403efe..6579fa3c 100644 --- a/src/main/java/kafdrop/service/KafkaMonitorImpl.java +++ b/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 getConsumerOffsets(Set topics) { .filter(not(ConsumerGroupOffsets::isEmpty)) .collect(Collectors.toList()); } + + @Override + public RecordMetadata publishMessage(CreateMessageVO message, Serializers serializers) { + return highLevelProducer.publishMessage(message, serializers); + } } diff --git a/src/main/java/kafdrop/service/KafkaProducerException.java b/src/main/java/kafdrop/service/KafkaProducerException.java new file mode 100644 index 00000000..06aadd52 --- /dev/null +++ b/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); + } +} diff --git a/src/main/java/kafdrop/util/AvroMessageSerializer.java b/src/main/java/kafdrop/util/AvroMessageSerializer.java new file mode 100644 index 00000000..2cb9b840 --- /dev/null +++ b/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(); + 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; + } + +} diff --git a/src/main/java/kafdrop/util/DefaultMessageSerializer.java b/src/main/java/kafdrop/util/DefaultMessageSerializer.java new file mode 100644 index 00000000..5cb3555b --- /dev/null +++ b/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(); + } + +} diff --git a/src/main/java/kafdrop/util/MessageSerializer.java b/src/main/java/kafdrop/util/MessageSerializer.java new file mode 100644 index 00000000..657be5c5 --- /dev/null +++ b/src/main/java/kafdrop/util/MessageSerializer.java @@ -0,0 +1,6 @@ +package kafdrop.util; + +@FunctionalInterface +public interface MessageSerializer { + byte[] serializeMessage(String value); +} \ No newline at end of file diff --git a/src/main/java/kafdrop/util/MsgPackMessageSerializer.java b/src/main/java/kafdrop/util/MsgPackMessageSerializer.java new file mode 100644 index 00000000..371cb284 --- /dev/null +++ b/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); + } + } + +} diff --git a/src/main/java/kafdrop/util/ProtobufMessageDeserializer.java b/src/main/java/kafdrop/util/ProtobufMessageDeserializer.java index 8eb9554d..39a0a8a2 100644 --- a/src/main/java/kafdrop/util/ProtobufMessageDeserializer.java +++ b/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); diff --git a/src/main/java/kafdrop/util/ProtobufMessageSerializer.java b/src/main/java/kafdrop/util/ProtobufMessageSerializer.java new file mode 100644 index 00000000..7befd812 --- /dev/null +++ b/src/main/java/kafdrop/util/ProtobufMessageSerializer.java @@ -0,0 +1,75 @@ +package kafdrop.util; + +import java.io.File; +import java.io.FileInputStream; +import java.io.FileNotFoundException; +import java.io.IOException; +import java.io.InputStream; +import java.util.ArrayList; +import java.util.List; +import java.util.stream.Collectors; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import com.google.protobuf.Descriptors; +import com.google.protobuf.DynamicMessage; +import com.google.protobuf.DescriptorProtos.FileDescriptorProto; +import com.google.protobuf.DescriptorProtos.FileDescriptorSet; +import com.google.protobuf.Descriptors.DescriptorValidationException; +import com.google.protobuf.Descriptors.FileDescriptor; + +public class ProtobufMessageSerializer implements MessageSerializer { + + private String fullDescFile; + private String msgTypeName; + + private static final Logger LOGGER = LoggerFactory.getLogger(ProtobufMessageSerializer.class); + + public ProtobufMessageSerializer(String fullDescFile, String msgTypeName) { + this.fullDescFile = fullDescFile; + this.msgTypeName = msgTypeName; + } + + @Override + public byte[] serializeMessage(String value) { + try (InputStream input = new FileInputStream(new File(fullDescFile))) { + FileDescriptorSet set = FileDescriptorSet.parseFrom(input); + + List fileDescriptors = new ArrayList<>(); + for (FileDescriptorProto ffdp : set.getFileList()) { + FileDescriptor fileDescriptor = Descriptors.FileDescriptor.buildFrom(ffdp, + (FileDescriptor[]) fileDescriptors.toArray(new FileDescriptor[fileDescriptors.size()])); + fileDescriptors.add(fileDescriptor); + } + + final var descriptors = fileDescriptors.stream().flatMap(desc -> desc.getMessageTypes().stream()) + .collect(Collectors.toList()); + + final var messageDescriptor = descriptors.stream().filter(desc -> msgTypeName.equals(desc.getName())) + .findFirst(); + if (messageDescriptor.isEmpty()) { + final String errorMsg = String.format("Can't find specific message type: %s", msgTypeName); + LOGGER.error(errorMsg); + throw new SerializationException(errorMsg); + } + + return DynamicMessage.parseFrom(messageDescriptor.get(), + value.getBytes()).toByteArray(); + + } catch (FileNotFoundException e) { + final String errorMsg = String.format("Couldn't open descriptor file: %s", fullDescFile); + LOGGER.error(errorMsg, e); + throw new SerializationException(errorMsg); + } catch (IOException e) { + final String errorMsg = "Can't decode Protobuf message"; + LOGGER.error(errorMsg, e); + throw new SerializationException(errorMsg); + } catch (DescriptorValidationException e) { + final String errorMsg = String.format("Can't compile proto message type: %s", msgTypeName); + LOGGER.error(errorMsg, e); + throw new SerializationException(errorMsg); + } + } + +} diff --git a/src/main/java/kafdrop/util/SerializationException.java b/src/main/java/kafdrop/util/SerializationException.java new file mode 100644 index 00000000..6021e07b --- /dev/null +++ b/src/main/java/kafdrop/util/SerializationException.java @@ -0,0 +1,14 @@ +package kafdrop.util; + +public class SerializationException extends RuntimeException { + + /** + * + */ + private static final long serialVersionUID = 1L; + + public SerializationException(String msg) { + super(msg); + } + +} diff --git a/src/main/java/kafdrop/util/Serializers.java b/src/main/java/kafdrop/util/Serializers.java new file mode 100644 index 00000000..e66fef71 --- /dev/null +++ b/src/main/java/kafdrop/util/Serializers.java @@ -0,0 +1,21 @@ +package kafdrop.util; + +public class Serializers { + + private final MessageSerializer keySerializer; + private final MessageSerializer valueSerializer; + + public Serializers(MessageSerializer keySerializer, MessageSerializer valueSerializer) { + this.keySerializer = keySerializer; + this.valueSerializer = valueSerializer; + } + + public MessageSerializer getKeySerializer() { + return keySerializer; + } + + public MessageSerializer getValueSerializer() { + return valueSerializer; + } + +} diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 23bdeb25..6701528a 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -1,36 +1,36 @@ -server: - port: ${SERVER_PORT:9000} - servlet: - context-path: / - error: - whitelabel: - enabled: false - -spring: - jmx: - enabled: true - default_domain: Kafdrop - jackson: - deserialization: - fail_on_unknown_properties: false - read_unknown_enum_values_as_null: true - -management: - endpoints: - web: - base-path: /actuator - exposure.include: "*" - server: - port: 9000 - -kafdrop.monitor: - clientId: Kafdrop - -kafka: - brokerConnect: localhost:9092 - isSecured: false - saslMechanism: "PLAIN" - securityProtocol: "SASL_PLAINTEXT" - truststoreFile: "${KAFKA_TRUSTSTORE_FILE:kafka.truststore.jks}" - propertiesFile : "${KAFKA_PROPERTIES_FILE:kafka.properties}" +server: + port: ${SERVER_PORT:9000} + servlet: + context-path: / + error: + whitelabel: + enabled: false + +spring: + jmx: + enabled: true + default_domain: Kafdrop + jackson: + deserialization: + fail_on_unknown_properties: false + read_unknown_enum_values_as_null: true + +management: + endpoints: + web: + base-path: /actuator + exposure.include: "*" + server: + port: 9000 + +kafdrop.monitor: + clientId: Kafdrop + +kafka: + brokerConnect: localhost:9092 + isSecured: false + saslMechanism: "PLAIN" + securityProtocol: "SASL_PLAINTEXT" + truststoreFile: "${KAFKA_TRUSTSTORE_FILE:kafka.truststore.jks}" + propertiesFile : "${KAFKA_PROPERTIES_FILE:kafka.properties}" keystoreFile: "${KAFKA_KEYSTORE_FILE:kafka.keystore.jks}" \ No newline at end of file diff --git a/src/main/resources/templates/message-inspector.ftl b/src/main/resources/templates/message-inspector.ftl index 2911f5a0..86ce5f9a 100644 --- a/src/main/resources/templates/message-inspector.ftl +++ b/src/main/resources/templates/message-inspector.ftl @@ -29,6 +29,11 @@ #partitionSizes { margin-left: 16px; } + + .top-nav-buttons{ + margin-left: 16px; + margin-top: 16px; + } .badge { margin-right: 5px; @@ -81,6 +86,56 @@ id="partitionSize">${curPartition.size - curPartition.firstOffset} + + + + + + + +
@@ -149,7 +204,7 @@
   - +