Add keyDecoder,maxRecords,timeoutMs parameters when getting records by offset. Add api to stop reassign partitions.

Author: tinawenqiao

src/main/java/org/gnuhpc/bigdata/controller/KafkaController.java

@@ -172,14 +172,18 @@ public boolean existTopic(@PathVariable String topic) {
   @GetMapping(value = "/consumer/{topic}/{partition}/{offset}")
   @ApiOperation(
       value =
-          "Get the message from the offset of the partition in the topic"
-              + ", decoder is not supported yet")
-  public Record getMessage(
+          "Get the message from the offset of the partition in the topic")
+  public List<Record> getMessage(
       @PathVariable String topic,
       @PathVariable int partition,
       @PathVariable long offset,
-      @RequestParam(required = false) String decoder) {
-    return kafkaAdminService.getRecordByOffset(topic, partition, offset, decoder, "");
+      @RequestParam(required = false) int maxRecords,
+      @RequestParam(required = false, defaultValue = "StringDeserializer") String keyDecoder,
+      @RequestParam(required = false, defaultValue = "StringDeserializer") String valueDecoder,
+      @RequestParam(required = false) String avroSchema,
+      @RequestParam(required = false, defaultValue = "30000") long fetchTimeoutMs) throws ApiException {
+    return kafkaAdminService.getRecordsByOffset(topic, partition, offset, maxRecords, keyDecoder,
+        valueDecoder, avroSchema, fetchTimeoutMs);
   }
 
   @GetMapping(value = "/topics/{topic}")
@@ -266,6 +270,12 @@ public ReassignStatus checkReassignPartitions(@RequestBody ReassignModel reassig
     return kafkaAdminService.checkReassignStatus(reassign);
   }
 
+  @PutMapping(value = "/partitions/reassign/stop")
+  @ApiOperation(value = "Stop the partition reassignment process")
+  public GeneralResponse stopReassignPartitions() {
+    return kafkaAdminService.stopReassignPartitions();
+  }
+
   @GetMapping(value = "/consumergroups")
   @ApiOperation(value = "List all consumer groups from zk and kafka")
   public Map<String, Set<String>> listAllConsumerGroups(

src/main/java/org/gnuhpc/bigdata/model/Record.java

@@ -5,23 +5,28 @@
 import lombok.Data;
 import lombok.Getter;
 import lombok.Setter;
+import lombok.extern.log4j.Log4j;
 import org.apache.kafka.common.errors.ApiException;
 import org.apache.kafka.common.utils.Bytes;
+import org.gnuhpc.bigdata.utils.KafkaUtils;
 
 @Data
 @Getter
 @Setter
 @Builder
+@Log4j
 public class Record {
   public String topic;
   public long offset;
   public Object key = new Object();
   public Object value = new Object();
   public long timestamp;
-  public Class<?> type;
-  String decoder;
+  String keyDecoder;
+  String valueDecoder;
 
-  public String getValue() {
+  public String getValueByDecoder(String decoder, Object value) {
+    if (value == null) return null;
+    Class<?> type = KafkaUtils.DESERIALIZER_TYPE_MAP.get(decoder);
     try {
       if (String.class.isAssignableFrom(type)) {
         return value.toString();
@@ -53,7 +58,7 @@ public String getValue() {
       }
 
       if (byte[].class.isAssignableFrom(type)) {
-        if (decoder.equals("AvorDeserializer")) {
+        if (decoder.equals("AvroDeserializer")) {
           return value.toString();
         } else {
           byte[] byteArray = (byte[]) value;
@@ -74,6 +79,16 @@ public String getValue() {
             + "String, Short, Integer, Long, Float, Double, ByteArray, ByteBuffer, Bytes");
   }
 
+  public String getValue() {
+    log.info("getValue for value:" + value + " by decoder:" + valueDecoder);
+    return getValueByDecoder(valueDecoder, value);
+  }
+
+  public String getKey() {
+    log.info("getKeyValue for key:" + key + " by decoder:" + keyDecoder);
+    return getValueByDecoder(keyDecoder, key);
+  }
+
   @Override
   public String toString() {
     if (value != null) {
@@ -82,13 +97,15 @@ public String toString() {
           + ", offset:"
           + offset
           + ", key:"
-          + key
+          + getKey()
           + ", value:"
           + getValue()
           + ", timestamp:"
           + timestamp
-          + ", type:"
-          + type;
+          + ", keyDecoder:"
+          + keyDecoder
+          + ", valueDecoder:"
+          + valueDecoder;
     } else {
       return "topic:"
           + topic
@@ -100,8 +117,10 @@ public String toString() {
           + value
           + ", timestamp:"
           + timestamp
-          + ", type:"
-          + type;
+          + ", keyDecoder:"
+          + keyDecoder
+          + ", valueDecoder:"
+          + valueDecoder;
     }
   }
 }