变更副本信息查询

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

@@ -0,0 +1,27 @@
+package com.xuxd.kafka.console.beans;
+
+import java.util.List;
+import lombok.Data;
+
+/**
+ * kafka-console-ui.
+ *
+ * @author xuxd
+ * @date 2021-11-19 17:07:50
+ **/
+@Data
+public class ReplicaAssignment {
+
+    private long version = 1L;
+
+    private List<Partition> partitions;
+
+    @Data
+    static class Partition {
+        private String topic;
+
+        private int partition;
+
+        private List<Integer> replicas;
+    }
+}

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

@@ -71,4 +71,8 @@ public class TopicController {
 
         return topicService.addPartitions(topic, addNum, assignment);
     }
+    @GetMapping("/replica/assignment")
+    public Object getCurrentReplicaAssignment(@RequestParam String topic) {
+        return topicService.getCurrentReplicaAssignment(topic);
+    }
 }

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

@@ -26,4 +26,6 @@ public interface TopicService {
     ResponseData createTopic(NewTopic topic);
 
     ResponseData addPartitions(String topic, int addNum, List<List<Integer>> newAssignmentst);
+
+    ResponseData getCurrentReplicaAssignment(String topic);
 }

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

@@ -8,6 +8,7 @@ import com.xuxd.kafka.console.beans.dos.MinOffsetAlignmentDO;
 import com.xuxd.kafka.console.beans.vo.OffsetAlignmentVO;
 import com.xuxd.kafka.console.dao.MinOffsetAlignmentMapper;
 import com.xuxd.kafka.console.service.OperationService;
+import com.xuxd.kafka.console.utils.GsonUtil;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -30,7 +31,7 @@ import scala.Tuple2;
 @Service
 public class OperationServiceImpl implements OperationService {
 
-    private Gson gson = new Gson();
+    private Gson gson = GsonUtil.INSTANCE.get();
 
     @Autowired
     private OperationConsole operationConsole;

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

@@ -1,10 +1,13 @@
 package com.xuxd.kafka.console.service.impl;
 
+import com.google.gson.Gson;
+import com.xuxd.kafka.console.beans.ReplicaAssignment;
 import com.xuxd.kafka.console.beans.ResponseData;
 import com.xuxd.kafka.console.beans.enums.TopicType;
 import com.xuxd.kafka.console.beans.vo.TopicDescriptionVO;
 import com.xuxd.kafka.console.beans.vo.TopicPartitionVO;
 import com.xuxd.kafka.console.service.TopicService;
+import com.xuxd.kafka.console.utils.GsonUtil;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
@@ -38,6 +41,8 @@ public class TopicServiceImpl implements TopicService {
     @Autowired
     private TopicConsole topicConsole;
 
+    private Gson gson = GsonUtil.INSTANCE.get();
+
     @Override public ResponseData getTopicNameList(boolean internal) {
         return ResponseData.create().data(topicConsole.getTopicNameList(internal)).success();
     }
@@ -130,4 +135,11 @@ public class TopicServiceImpl implements TopicService {
 
         return success ? ResponseData.create().success() : ResponseData.create().failed(tuple2._2());
     }
+
+    @Override public ResponseData getCurrentReplicaAssignment(String topic) {
+        Tuple2<Object, String> tuple2 = topicConsole.getCurrentReplicaAssignmentJson(topic);
+        boolean success = (boolean) tuple2._1();
+
+        return success ? ResponseData.create().data(gson.fromJson(tuple2._2(), ReplicaAssignment.class)).success() : ResponseData.create().failed(tuple2._2());
+    }
 }

src/main/java/com/xuxd/kafka/console/utils/GsonUtil.java

@@ -0,0 +1,19 @@
+package com.xuxd.kafka.console.utils;
+
+import com.google.gson.Gson;
+
+/**
+ * kafka-console-ui.
+ *
+ * @author xuxd
+ * @date 2021-11-19 17:01:01
+ **/
+public enum GsonUtil {
+    INSTANCE;
+
+    private Gson gson = new Gson();
+
+    public Gson get() {
+        return gson;
+    }
+}

src/main/scala/kafka/console/TopicConsole.scala

@@ -1,14 +1,17 @@
 package kafka.console
 
-import java.util
-import java.util.concurrent.TimeUnit
-import java.util.{Collections, List, Set}
-
 import com.xuxd.kafka.console.config.KafkaConfig
+import kafka.admin.ReassignPartitionsCommand.compareTopicPartitions
+import kafka.utils.Json
 import org.apache.kafka.clients.admin._
-import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.errors.UnknownTopicOrPartitionException
+import org.apache.kafka.common.{TopicPartition, TopicPartitionReplica}
 
-import scala.jdk.CollectionConverters.{CollectionHasAsScala, SetHasAsJava}
+import java.util
+import java.util.concurrent.{ExecutionException, TimeUnit}
+import java.util.{Collections, List, Set}
+import scala.collection.{Map, Seq}
+import scala.jdk.CollectionConverters.{CollectionHasAsScala, MapHasAsJava, MapHasAsScala, SeqHasAsJava, SetHasAsJava}
 
 /**
  * kafka-console-ui.
@@ -121,4 +124,61 @@ class TopicConsole(config: KafkaConfig) extends KafkaConsole(config: KafkaConfig
             (false, e.getMessage)
         }).asInstanceOf[(Boolean, String)]
     }
+
+    def getCurrentReplicaAssignmentJson(topic: String): (Boolean, String) = {
+        withAdminClientAndCatchError(admin => {
+            val json = formatAsReassignmentJson(getReplicaAssignmentForTopics(admin, Seq(topic)), Map.empty)
+            (true, json)
+        }, e => {
+            log.error("getCurrentReplicaAssignmentJson error, ", e)
+            (false, e.getMessage)
+        }).asInstanceOf[(Boolean, String)]
+    }
+
+    /**
+     * Get the current replica assignments for some topics.
+     *
+     * @param adminClient The AdminClient to use.
+     * @param topics      The topics to get information about.
+     * @return A map from partitions to broker assignments.
+     *         If any topic can't be found, an exception will be thrown.
+     */
+    private def getReplicaAssignmentForTopics(adminClient: Admin,
+        topics: Seq[String])
+    : Map[TopicPartition, Seq[Int]] = {
+        describeTopics(adminClient, topics.toSet.asJava).flatMap {
+            case (topicName, topicDescription) => topicDescription.partitions.asScala.map { info =>
+                (new TopicPartition(topicName, info.partition), info.replicas.asScala.map(_.id).toSeq)
+            }
+        }
+    }
+
+    private def formatAsReassignmentJson(partitionsToBeReassigned: Map[TopicPartition, Seq[Int]],
+        replicaLogDirAssignment: Map[TopicPartitionReplica, String]): String = {
+        Json.encodeAsString(Map(
+            "version" -> 1,
+            "partitions" -> partitionsToBeReassigned.keySet.toBuffer.sortWith(compareTopicPartitions).map {
+                tp =>
+                    val replicas = partitionsToBeReassigned(tp)
+                    Map(
+                        "topic" -> tp.topic,
+                        "partition" -> tp.partition,
+                        "replicas" -> replicas.asJava
+                    ).asJava
+            }.asJava
+        ).asJava)
+    }
+
+    private def describeTopics(adminClient: Admin,
+        topics: Set[String])
+    : Map[String, TopicDescription] = {
+        adminClient.describeTopics(topics).values.asScala.map { case (topicName, topicDescriptionFuture) =>
+            try topicName -> topicDescriptionFuture.get
+            catch {
+                case t: ExecutionException if t.getCause.isInstanceOf[UnknownTopicOrPartitionException] =>
+                    throw new ExecutionException(
+                        new UnknownTopicOrPartitionException(s"Topic $topicName not found."))
+            }
+        }
+    }
 }