查询消息详情信息

src/main/java/com/xuxd/kafka/console/beans/QueryMessage.java

@@ -20,4 +20,8 @@ public class QueryMessage {
     private long endTime;
 
     private long offset;
+
+    private String keyDeserializer;
+
+    private String valueDeserializer;
 }

src/main/java/com/xuxd/kafka/console/beans/dto/QueryMessageDTO.java

@@ -23,6 +23,10 @@ public class QueryMessageDTO {
 
     private Long offset;
 
+    private String keyDeserializer;
+
+    private String valueDeserializer;
+
     public QueryMessage toQueryMessage() {
         QueryMessage queryMessage = new QueryMessage();
         queryMessage.setTopic(topic);
@@ -38,6 +42,9 @@ public class QueryMessageDTO {
             queryMessage.setOffset(offset);
         }
 
+        queryMessage.setKeyDeserializer(keyDeserializer);
+        queryMessage.setValueDeserializer(valueDeserializer);
+
         return queryMessage;
     }
 }

src/main/java/com/xuxd/kafka/console/beans/vo/MessageDetailVO.java

@@ -0,0 +1,38 @@
+package com.xuxd.kafka.console.beans.vo;
+
+import java.util.ArrayList;
+import java.util.List;
+import lombok.Data;
+
+/**
+ * kafka-console-ui.
+ *
+ * @author xuxd
+ * @date 2021-12-12 12:45:23
+ **/
+@Data
+public class MessageDetailVO {
+
+    private String topic;
+
+    private int partition;
+
+    private long offset;
+
+    private long timestamp;
+
+    private String timestampType;
+
+    private List<HeaderVO> headers = new ArrayList<>();
+
+    private Object key;
+
+    private Object value;
+
+    @Data
+    public static class HeaderVO {
+        String key;
+
+        String value;
+    }
+}

src/main/java/com/xuxd/kafka/console/controller/MessageController.java

@@ -3,6 +3,7 @@ package com.xuxd.kafka.console.controller;
 import com.xuxd.kafka.console.beans.dto.QueryMessageDTO;
 import com.xuxd.kafka.console.service.MessageService;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 import org.springframework.web.bind.annotation.RequestMapping;
@@ -30,4 +31,14 @@ public class MessageController {
     public Object searchByOffset(@RequestBody QueryMessageDTO dto) {
         return messageService.searchByOffset(dto.toQueryMessage());
     }
+
+    @PostMapping("/search/detail")
+    public Object searchDetail(@RequestBody QueryMessageDTO dto) {
+        return messageService.searchDetail(dto.toQueryMessage());
+    }
+
+    @GetMapping("/deserializer/list")
+    public Object deserializerList() {
+        return messageService.deserializerList();
+    }
 }

src/main/java/com/xuxd/kafka/console/service/MessageService.java

@@ -14,4 +14,8 @@ public interface MessageService {
     ResponseData searchByTime(QueryMessage queryMessage);
 
     ResponseData searchByOffset(QueryMessage queryMessage);
+
+    ResponseData searchDetail(QueryMessage queryMessage);
+
+    ResponseData deserializerList();
 }

src/main/java/com/xuxd/kafka/console/service/impl/MessageServiceImpl.java

@@ -3,6 +3,7 @@ package com.xuxd.kafka.console.service.impl;
 import com.xuxd.kafka.console.beans.QueryMessage;
 import com.xuxd.kafka.console.beans.ResponseData;
 import com.xuxd.kafka.console.beans.vo.ConsumerRecordVO;
+import com.xuxd.kafka.console.beans.vo.MessageDetailVO;
 import com.xuxd.kafka.console.service.MessageService;
 import java.util.Collections;
 import java.util.HashMap;
@@ -14,9 +15,15 @@ import java.util.stream.Collectors;
 import kafka.console.MessageConsole;
 import kafka.console.TopicConsole;
 import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.kafka.clients.admin.TopicDescription;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.DoubleDeserializer;
+import org.apache.kafka.common.serialization.FloatDeserializer;
+import org.apache.kafka.common.serialization.IntegerDeserializer;
+import org.apache.kafka.common.serialization.StringDeserializer;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
@@ -35,6 +42,17 @@ public class MessageServiceImpl implements MessageService {
     @Autowired
     private TopicConsole topicConsole;
 
+    private Map<String, Deserializer> deserializerDict = new HashMap<>();
+
+    {
+        deserializerDict.put("Integer", new IntegerDeserializer());
+        deserializerDict.put("String", new StringDeserializer());
+        deserializerDict.put("Float", new FloatDeserializer());
+        deserializerDict.put("Double", new DoubleDeserializer());
+    }
+
+    public static String defaultDeserializer = "String";
+
     @Override public ResponseData searchByTime(QueryMessage queryMessage) {
         int maxNums = 10000;
 
@@ -50,14 +68,68 @@ public class MessageServiceImpl implements MessageService {
     }
 
     @Override public ResponseData searchByOffset(QueryMessage queryMessage) {
+        Map<TopicPartition, ConsumerRecord<byte[], byte[]>> recordMap = searchRecordByOffset(queryMessage);
+
+        return ResponseData.create().data(recordMap.values().stream().map(ConsumerRecordVO::fromConsumerRecord).collect(Collectors.toList())).success();
+    }
+
+    @Override public ResponseData searchDetail(QueryMessage queryMessage) {
+        if (queryMessage.getPartition() == -1) {
+            throw new IllegalArgumentException();
+        }
+        if (StringUtils.isBlank(queryMessage.getKeyDeserializer())) {
+            queryMessage.setKeyDeserializer(defaultDeserializer);
+        }
+
+        if (StringUtils.isBlank(queryMessage.getValueDeserializer())) {
+            queryMessage.setValueDeserializer(defaultDeserializer);
+        }
+
+        Map<TopicPartition, ConsumerRecord<byte[], byte[]>> recordMap = searchRecordByOffset(queryMessage);
+        ConsumerRecord<byte[], byte[]> record = recordMap.get(new TopicPartition(queryMessage.getTopic(), queryMessage.getPartition()));
+        if (record != null) {
+            MessageDetailVO vo = new MessageDetailVO();
+            vo.setTopic(record.topic());
+            vo.setPartition(record.partition());
+            vo.setOffset(record.offset());
+            vo.setTimestamp(record.timestamp());
+            vo.setTimestampType(record.timestampType().name());
+            try {
+                vo.setKey(deserializerDict.get(queryMessage.getKeyDeserializer()).deserialize(queryMessage.getTopic(), record.key()));
+            } catch (Exception e) {
+                vo.setKey("KeyDeserializer Error: " + e.getMessage());
+            }
+            try {
+                vo.setValue(deserializerDict.get(queryMessage.getValueDeserializer()).deserialize(queryMessage.getTopic(), record.value()));
+            } catch (Exception e) {
+                vo.setValue("ValueDeserializer Error: " + e.getMessage());
+            }
+
+            record.headers().forEach(header -> {
+                MessageDetailVO.HeaderVO headerVO = new MessageDetailVO.HeaderVO();
+                headerVO.setKey(header.key());
+                headerVO.setValue(new String(header.value()));
+                vo.getHeaders().add(headerVO);
+            });
+
+            return ResponseData.create().data(vo).success();
+        }
+        return ResponseData.create().failed("Not found message detail.");
+    }
+
+    @Override public ResponseData deserializerList() {
+        return ResponseData.create().data(deserializerDict.keySet()).success();
+    }
+
+    private Map<TopicPartition, ConsumerRecord<byte[], byte[]>> searchRecordByOffset(QueryMessage queryMessage) {
         Set<TopicPartition> partitions = getPartitions(queryMessage);
+
         Map<TopicPartition, Object> offsetTable = new HashMap<>();
         partitions.forEach(tp -> {
             offsetTable.put(tp, queryMessage.getOffset());
         });
         Map<TopicPartition, ConsumerRecord<byte[], byte[]>> recordMap = messageConsole.searchBy(offsetTable);
-
-        return ResponseData.create().data(recordMap.values().stream().map(ConsumerRecordVO::fromConsumerRecord).collect(Collectors.toList())).success();
+        return recordMap;
     }
 
     private Set<TopicPartition> getPartitions(QueryMessage queryMessage) {