From 98e35cbee7ceeadd74d6f20b68563ae8aaf6adfa Mon Sep 17 00:00:00 2001 From: Faaez Ahmed Chishti Date: Thu, 26 Nov 2020 03:36:38 -0600 Subject: [PATCH 1/3] Added view all messages button in message inspector page --- src/main/resources/templates/message-inspector.ftl | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/src/main/resources/templates/message-inspector.ftl b/src/main/resources/templates/message-inspector.ftl index 2911f5a0..0ff63e38 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; } + + #allMessages{ + margin-left: 16px; + margin-top: 16px; + } .badge { margin-right: 5px; @@ -81,6 +86,12 @@ id="partitionSize">${curPartition.size - curPartition.firstOffset} +
+ + All Messages + +
+
@@ -149,7 +160,7 @@
   - +
From f096ac34437d9640b38cd30f85d0e325435ff4f0 Mon Sep 17 00:00:00 2001 From: Faaez Ahmed Chishti Date: Tue, 29 Dec 2020 23:34:30 -0600 Subject: [PATCH 2/3] Added functionality to allow user to add message to a topic partition manually --- kaas_local_jaas.conf | 6 +- .../kafdrop/controller/MessageController.java | 78 +++++++++++++++++++ .../java/kafdrop/model/CreateMessageVO.java | 24 ++++++ .../service/KafkaHighLevelProducer.java | 78 +++++++++++++++++++ .../java/kafdrop/service/KafkaMonitor.java | 4 + .../kafdrop/service/KafkaMonitorImpl.java | 11 ++- .../service/KafkaProducerException.java | 16 ++++ .../kafdrop/util/AvroMessageSerializer.java | 36 +++++++++ .../util/DefaultMessageSerializer.java | 10 +++ .../java/kafdrop/util/MessageSerializer.java | 6 ++ .../util/MsgPackMessageSerializer.java | 26 +++++++ .../util/ProtobufMessageDeserializer.java | 2 +- .../util/ProtobufMessageSerializer.java | 75 ++++++++++++++++++ .../kafdrop/util/SerializationException.java | 14 ++++ src/main/java/kafdrop/util/Serializers.java | 21 +++++ src/main/resources/application.yml | 71 ++++++++--------- .../resources/templates/message-inspector.ftl | 54 +++++++++++-- 17 files changed, 487 insertions(+), 45 deletions(-) create mode 100644 src/main/java/kafdrop/model/CreateMessageVO.java create mode 100644 src/main/java/kafdrop/service/KafkaHighLevelProducer.java create mode 100644 src/main/java/kafdrop/service/KafkaProducerException.java create mode 100644 src/main/java/kafdrop/util/AvroMessageSerializer.java create mode 100644 src/main/java/kafdrop/util/DefaultMessageSerializer.java create mode 100644 src/main/java/kafdrop/util/MessageSerializer.java create mode 100644 src/main/java/kafdrop/util/MsgPackMessageSerializer.java create mode 100644 src/main/java/kafdrop/util/ProtobufMessageSerializer.java create mode 100644 src/main/java/kafdrop/util/SerializationException.java create mode 100644 src/main/java/kafdrop/util/Serializers.java diff --git a/kaas_local_jaas.conf b/kaas_local_jaas.conf index 6dc2616b..40f6db3c 100644 --- a/kaas_local_jaas.conf +++ b/kaas_local_jaas.conf @@ -5,7 +5,7 @@ Client { }; KafkaClient { - org.apache.kafka.common.security.plain.PlainLoginModule required - username="xyz" - password="xyz"; + org.apache.kafka.common.security.scram.ScramLoginModule required + username="n2ls3q6r" + password="0F-UcfcAX25wmigKji4AYLFAR6kOqKzf"; }; \ No newline at end of file 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..91e4af67 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -1,36 +1,37 @@ -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: tricycle-01.srvs.cloudkafka.com:9094,tricycle-02.srvs.cloudkafka.com:9094,tricycle-03.srvs.cloudkafka.com:9094 + isSecured: true + env: local + saslMechanism: SCRAM-SHA-256 + securityProtocol: SASL_SSL + 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 0ff63e38..86ce5f9a 100644 --- a/src/main/resources/templates/message-inspector.ftl +++ b/src/main/resources/templates/message-inspector.ftl @@ -30,7 +30,7 @@ margin-left: 16px; } - #allMessages{ + .top-nav-buttons{ margin-left: 16px; margin-top: 16px; } @@ -86,11 +86,55 @@ id="partitionSize">${curPartition.size - curPartition.firstOffset} -
- - All Messages - + + + + + + +
From f0ba1d54d89d41a5de6d44efa50c27e8601631ec Mon Sep 17 00:00:00 2001 From: Faaez Ahmed Chishti Date: Wed, 30 Dec 2020 00:02:27 -0600 Subject: [PATCH 3/3] Reverted config files --- kaas_local_jaas.conf | 6 +++--- src/main/resources/application.yml | 9 ++++----- 2 files changed, 7 insertions(+), 8 deletions(-) diff --git a/kaas_local_jaas.conf b/kaas_local_jaas.conf index 40f6db3c..6dc2616b 100644 --- a/kaas_local_jaas.conf +++ b/kaas_local_jaas.conf @@ -5,7 +5,7 @@ Client { }; KafkaClient { - org.apache.kafka.common.security.scram.ScramLoginModule required - username="n2ls3q6r" - password="0F-UcfcAX25wmigKji4AYLFAR6kOqKzf"; + org.apache.kafka.common.security.plain.PlainLoginModule required + username="xyz" + password="xyz"; }; \ No newline at end of file diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml index 91e4af67..6701528a 100644 --- a/src/main/resources/application.yml +++ b/src/main/resources/application.yml @@ -27,11 +27,10 @@ kafdrop.monitor: clientId: Kafdrop kafka: - brokerConnect: tricycle-01.srvs.cloudkafka.com:9094,tricycle-02.srvs.cloudkafka.com:9094,tricycle-03.srvs.cloudkafka.com:9094 - isSecured: true - env: local - saslMechanism: SCRAM-SHA-256 - securityProtocol: SASL_SSL + 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