@Configuration
public class MessageFormatConfiguration {

   @Component
   @ConfigurationProperties(prefix = "message")
   public static class MessageFormatProperties
   {

      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;
      }

   }

} @Configuration
public class SchemaRegistryConfiguration {

   @Component
   @ConfigurationProperties(prefix = "schemaregistry")
   public static class SchemaRegistryProperties
   {

      public static final Pattern CONNECT_SEPARATOR = Pattern.compile("\\s*,\\s*");

      private String connect;

      public String getConnect()
      {
         return connect;
      }

      public void setConnect(String connect)
      {
         this.connect = connect;
      }

      public List<String> getConnectList()
      {
         return CONNECT_SEPARATOR.splitAsStream(this.connect)
               .map(String::trim)
               .filter(s -> s.length() > 0)
               .collect(Collectors.toList());
      }

   }

} public class MessageVO
{
   private String message;
   private String key;
   private String topic;
   private int partition;
   private long offset;
   private boolean valid = true;
   private long timestamp;
   private String timestampType;
   private final Map<String, String> headers = new TreeMap<>();

   public void addHeader(String key, String value)
   {
      headers.put(key, value);
   }

   public Map<String, String> getHeaders()
   {
      return headers;
   }

   public boolean isValid()
   {
      return valid;
   }

   public void setValid(boolean valid)
   {
      this.valid = valid;
   }

   public String getTopic()
   {
      return topic;
   }

   public void setTopic(String topic)
   {
      this.topic = topic;
   }

   public int getPartition()
   {
      return partition;
   }

   public void setPartition(int partition)
   {
      this.partition = partition;
   }

   public long getOffset()
   {
      return offset;
   }

   public void setOffset(long offset)
   {
      this.offset = offset;
   }

   public String getMessage()
   {
      return message;
   }

   public void setMessage(String message)
   {
      this.message = message;
   }

   public String getKey()
   {
      return key;
   }

   public void setKey(String key)
   {
      this.key = key;
   }

   public long getTimestamp()
   {
      return timestamp;
   }

   public void setTimestamp(long timestamp)
   {
      this.timestamp = timestamp;
   }

   public String getTimestampType()
   {
      return timestampType;
   }

   public void setTimestampType(String timestampType)
   {
      this.timestampType = timestampType;
   }
} public class TopicVO implements Comparable<TopicVO>
{
   private String name;
   private Map<Integer, TopicPartitionVO> partitions = new TreeMap<>();
   private Map<String, Object> config = new TreeMap<>();
   // description?
   // partition state
   // delete supported?


   public TopicVO(String name)
   {
      this.name = name;
   }

   public String getName()
   {
      return name;
   }

   public void setName(String name)
   {
      this.name = name;
   }

   public Map<String, Object> getConfig()
   {
      return config;
   }

   public void setConfig(Map<String, Object> config)
   {
      this.config = config;
   }

   public Collection<TopicPartitionVO> getPartitions()
   {
      return partitions.values();
   }

   public Optional<TopicPartitionVO> getPartition(int partitionId)
   {
      return Optional.ofNullable(partitions.get(partitionId));
   }

   public Map<Integer, TopicPartitionVO> getPartitionMap() {
     return Collections.unmodifiableMap(partitions);
   }

   public Collection<TopicPartitionVO> getLeaderPartitions(int brokerId)
   {
      return partitions.values().stream()
         .filter(tp -> tp.getLeader() != null && tp.getLeader().getId() == brokerId)
         .collect(Collectors.toList());
   }

   public Collection<TopicPartitionVO> getUnderReplicatedPartitions()
   {
      return getUnderReplicatedPartitionStream()
         .collect(Collectors.toList());
   }

   private Stream<TopicPartitionVO> getUnderReplicatedPartitionStream()
   {
      return partitions.values().stream()
         .filter(TopicPartitionVO::isUnderReplicated);
   }

   /**
    * Returns the brokers that have under-replicated partition data
    * @return Set of broker ids
    */
   public Set<Integer> getUnderreplicatedBrokerIds()
   {
      return getUnderReplicatedPartitionStream()
         .flatMap(TopicPartitionVO::getOutOfSyncReplicaStream)
         .collect(Collectors.toSet());
   }

   public void setPartitions(Map<Integer, TopicPartitionVO> partitions)
   {
      this.partitions = partitions;
   }

   /**
    * Returns the total number of messages published to the topic, ever
    * @return
    */
   public long getTotalSize()
   {
      return partitions.values().stream()
         .map(TopicPartitionVO::getSize)
         .reduce(0L, Long::sum);
   }

   /**
    * Returns the total number of messages available to consume from the topic.
    * @return
    */
   public long getAvailableSize()
   {
      return partitions.values().stream()
         .map(p -> p.getSize() - p.getFirstOffset())
         .reduce(0L, Long::sum);
   }

   public double getPreferredReplicaPercent()
   {
      long preferredLeaderCount = partitions.values().stream()
         .filter(TopicPartitionVO::isLeaderPreferred)
         .count();
      return ((double) preferredLeaderCount) / ((double)partitions.size());
   }

   public Map<String, List<TopicPartitionVO.PartitionReplica>> getBrokerLeaders()
   {
      return partitions.values().stream()
              .map(TopicPartitionVO::getLeader)
              .filter(Objects::nonNull)
              .collect(Collectors.groupingBy(leader -> leader.getId().toString(), TreeMap::new, Collectors.toList()));
   }

   public Map<Integer, BrokerReplicas> getBrokerReplicas()
   {
      Map<Integer, BrokerReplicas> brokerReplicasMap = new TreeMap<>();
      partitions.values().stream()
         .flatMap(tp -> tp.getReplicas().stream())
         .forEach(r -> brokerReplicasMap.computeIfAbsent(r.getId(), id -> new BrokerReplicas(id.toString()))
            .addReplica(r));
      return brokerReplicasMap;
   }

   public static class BrokerReplicas
   {
      /** The ID of the broker owning the replicas */
      private final String brokerId;
      /** List of leader replicas on this broker */
      private final List<TopicPartitionVO.PartitionReplica> leaders = new ArrayList<>();
      /** List of non-leader replicas on this broker */
      private final List<TopicPartitionVO.PartitionReplica> replicas = new ArrayList<>();

      public BrokerReplicas(String brokerId) {
         this.brokerId = brokerId;
      }

      public void addReplica(TopicPartitionVO.PartitionReplica replica)
      {
         if (replica.isLeader()) {
            leaders.add(replica);
         }
         else {
            replicas.add(replica);
         }
      }

      public String getBrokerId()
      {
         return brokerId;
      }

      public List<TopicPartitionVO.PartitionReplica> getLeaders()
      {
         return leaders;
      }

      public List<TopicPartitionVO.PartitionReplica> getReplicas()
      {
         return replicas;
      }
   }

   public void addPartition(TopicPartitionVO partition)
   {
      partitions.put(partition.getId(), partition);
   }

   @Override
   public int compareTo(TopicVO that)
   {
      return this.name.compareTo(that.name);
   }

   @Override
   public boolean equals(Object o)
   {
      if (this == o) return true;
      if (o == null || getClass() != o.getClass()) return false;

      TopicVO that = (TopicVO) o;

      if (!name.equals(that.name)) return false;

      return true;
   }

   @Override
   public int hashCode()
   {
      return name.hashCode();
   }

} public interface KafkaMonitor
{
   List<BrokerVO> getBrokers();

   Optional<BrokerVO> getBroker(int id);

   default Optional<BrokerConfigurationVO> getBrokerConfiguration(int id)
   {
      return Optional.empty();
   }

   List<TopicVO> getTopics();

   Optional<TopicVO> getTopic(String topic);

   ClusterSummaryVO getClusterSummary(Collection<TopicVO> topics);

   List<ConsumerVO> getConsumers(TopicVO topic);

   Optional<ConsumerVO> getConsumer(String groupId);

   Optional<ConsumerVO> getConsumerByTopic(String groupId, TopicVO topic);
} @Service
public class MessageInspector
{
   private final Logger LOG = LoggerFactory.getLogger(getClass());

   private KafkaConfiguration config;

   public MessageInspector(KafkaConfiguration config)
   {
      this.config = config;
   }

   private Consumer<byte[], byte[]> createConsumer()
   {
      Properties properties = new Properties();
      config.applyCommon(properties);
      return new KafkaConsumer(properties, new ByteArrayDeserializer(), new ByteArrayDeserializer());

   }

   public List<MessageVO> getMessages(
           String topicName,
           int partitionId,
           long offset,
           long count,
           MessageDeserializer deserializer)
   {
      try (Consumer<byte[], byte[]> consumer = createConsumer())
      {
         TopicPartition topicPartition = new TopicPartition(topicName, partitionId);
         consumer.assign(Collections.singletonList(topicPartition));
         consumer.seek(topicPartition, offset);

         // Need to adjust how many records we read based on how many are actually available in the partition.
         long maxOffset = consumer.endOffsets(Collections.singletonList(topicPartition)).get(topicPartition);

         int numRecordsToRead = (int) Math.min(count, maxOffset - offset);

         List<MessageVO> recordList = new ArrayList<>(numRecordsToRead);
         while (recordList.size() < numRecordsToRead)
         {
            ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofMillis(1000L));
            List<ConsumerRecord<byte[], byte[]>> returnedRecords = records.records(topicPartition);

            returnedRecords.subList(0, Math.min((int) count - recordList.size(), returnedRecords.size()))
               .stream()
               .map(record -> createMessage(record, deserializer))
               .forEach(recordList::add);
         }

         return recordList;
      }
   }
   private MessageVO createMessage(ConsumerRecord<byte[], byte[]> record, MessageDeserializer deserializer)
   {
      MessageVO vo = new MessageVO();
      vo.setTopic(record.topic());
      vo.setOffset(record.offset());
      vo.setPartition(record.partition());
      if (record.key() != null && record.key().length > 0)
      {
         vo.setKey(readString(record.key()));
      }
      if (record.value() != null && record.value().length > 0)
      {
         final String messageString;
         if (deserializer != null)
         {
            messageString = deserializer.deserializeMessage(ByteBuffer.wrap(record.value()));
         }
         else
         {
            messageString = readString(record.value());
         }
         vo.setMessage(messageString);
      }

      vo.setTimestamp(record.timestamp());
      vo.setTimestampType(record.timestampType().toString());

      StreamSupport.stream(record.headers().spliterator(), false)
         .forEachOrdered(header -> vo.addHeader(header.key(), new String(header.value(), StandardCharsets.UTF_8)));

      return vo;
   }


} @ResponseStatus(HttpStatus.NOT_FOUND)
public class TopicNotFoundException extends RuntimeException
{
   public TopicNotFoundException()
   {
   }

   public TopicNotFoundException(String message)
   {
      super(message);
   }
} public class AvroMessageDeserializer implements MessageDeserializer {

   private String topicName;
   private String schemaRegistryUrl;

   public AvroMessageDeserializer(String topicName, String schemaRegistryUrl) {
      this.topicName = topicName;
      this.schemaRegistryUrl = schemaRegistryUrl;
   }

   @Override
   public String deserializeMessage(ByteBuffer buffer) {
      KafkaAvroDeserializer deserializer = getDeserializer();

      // Convert byte buffer to byte array
      byte[] bytes = ByteUtils.convertToByteArray(buffer);

      return formatJsonMessage(deserializer.deserialize(topicName, bytes).toString());
   }

   private String formatJsonMessage(String jsonMessage) {
      Gson gson = new GsonBuilder().setPrettyPrinting().create();
      JsonParser parser = new JsonParser();
      JsonElement element = parser.parse(jsonMessage);
      String formattedJsonMessage = gson.toJson(element);
      return formattedJsonMessage;
   }

   private KafkaAvroDeserializer getDeserializer() {
      Map<String, Object> config = new HashMap<>();
      config.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
      KafkaAvroDeserializer kafkaAvroDeserializer = new KafkaAvroDeserializer();
      kafkaAvroDeserializer.configure(config, false);
      return kafkaAvroDeserializer;
   }

} public class DefaultMessageDeserializer implements MessageDeserializer {

   @Override
   public String deserializeMessage(ByteBuffer buffer) {
      return ByteUtils.readString(buffer);
   }

} public interface MessageDeserializer {

   public String deserializeMessage(ByteBuffer buffer);

} public enum MessageFormat {
   DEFAULT, AVRO;
}
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.homeadvisor.kafdrop.config.MessageFormatConfiguration;
import com.homeadvisor.kafdrop.config.SchemaRegistryConfiguration;
import com.homeadvisor.kafdrop.model.MessageVO;
import com.homeadvisor.kafdrop.model.TopicVO;
import com.homeadvisor.kafdrop.service.KafkaMonitor;
import com.homeadvisor.kafdrop.service.MessageInspector;
import com.homeadvisor.kafdrop.service.TopicNotFoundException;
import com.homeadvisor.kafdrop.util.AvroMessageDeserializer;
import com.homeadvisor.kafdrop.util.DefaultMessageDeserializer;
import com.homeadvisor.kafdrop.util.MessageDeserializer;
import com.homeadvisor.kafdrop.util.MessageFormat;
import io.swagger.annotations.ApiOperation;
import io.swagger.annotations.ApiResponse;
import io.swagger.annotations.ApiResponses;
import org.springframework.beans.factory.annotation.Autowired;
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.*;
import javax.validation.Valid;
import javax.validation.constraints.Max;
import javax.validation.constraints.Min;
import javax.validation.constraints.NotNull;
import java.util.ArrayList;
import java.util.List;
/*
 * Copyright 2017 HomeAdvisor, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 *
 */

package com.homeadvisor.kafdrop.controller;




@Controller
public class MessageController
{
   @Autowired
   private KafkaMonitor kafkaMonitor;

   @Autowired
   private MessageInspector messageInspector;

   @Autowired
   private MessageFormatConfiguration.MessageFormatProperties messageFormatProperties;