preferred as leader.

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

@@ -0,0 +1,17 @@
+package com.xuxd.kafka.console.beans.dto;
+
+import lombok.Data;
+
+/**
+ * kafka-console-ui.
+ *
+ * @author xuxd
+ * @date 2021-11-10 20:09:20
+ **/
+@Data
+public class ReplicationDTO {
+
+    private String topic;
+
+    private int partition;
+}

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

@@ -1,5 +1,6 @@
 package com.xuxd.kafka.console.controller;
 
+import com.xuxd.kafka.console.beans.dto.ReplicationDTO;
 import com.xuxd.kafka.console.beans.dto.SyncDataDTO;
 import com.xuxd.kafka.console.service.OperationService;
 import org.apache.kafka.clients.admin.AdminClientConfig;
@@ -46,4 +47,9 @@ public class OperationController {
     public Object deleteAlignment(@RequestParam Long id) {
         return operationService.deleteAlignmentById(id);
     }
+
+    @PostMapping("/replication/preferred")
+    public Object electPreferredLeader(@RequestBody ReplicationDTO dto) {
+        return operationService.electPreferredLeader(dto.getTopic(), dto.getPartition());
+    }
 }

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

@@ -18,4 +18,6 @@ public interface OperationService {
     ResponseData getAlignmentList();
 
     ResponseData deleteAlignmentById(Long id);
+
+    ResponseData electPreferredLeader(String topic, int partition);
 }

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

@@ -9,9 +9,11 @@ import com.xuxd.kafka.console.beans.vo.OffsetAlignmentVO;
 import com.xuxd.kafka.console.dao.MinOffsetAlignmentMapper;
 import com.xuxd.kafka.console.service.OperationService;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
+import java.util.Set;
 import kafka.console.OperationConsole;
 import org.apache.kafka.common.TopicPartition;
 import org.springframework.beans.factory.ObjectProvider;
@@ -107,4 +109,17 @@ public class OperationServiceImpl implements OperationService {
         minOffsetAlignmentMapper.deleteById(id);
         return ResponseData.create().success();
     }
+
+    @Override public ResponseData electPreferredLeader(String topic, int partition) {
+        Set<TopicPartition> partitions = new HashSet<>();
+        if (partition != -1) {
+            partitions.add(new TopicPartition(topic, partition));
+        } else {
+
+            partitions.addAll(operationConsole.getTopicPartitions(topic));
+        }
+        Tuple2<Object, String> tuple2 = operationConsole.electPreferredLeader(partitions);
+
+        return (boolean) tuple2._1() ? ResponseData.create().success() : ResponseData.create().failed(tuple2._2());
+    }
 }

src/main/scala/kafka/console/KafkaConsole.scala

@@ -5,7 +5,7 @@ import java.util.Properties
 import com.xuxd.kafka.console.config.KafkaConfig
 import kafka.zk.{AdminZkClient, KafkaZkClient}
 import org.apache.kafka.clients.CommonClientConfigs
-import org.apache.kafka.clients.admin.{Admin, AdminClientConfig}
+import org.apache.kafka.clients.admin.{AbstractOptions, Admin, AdminClientConfig}
 import org.apache.kafka.clients.consumer.{ConsumerConfig, KafkaConsumer}
 import org.apache.kafka.common.config.SaslConfigs
 import org.apache.kafka.common.serialization.ByteArrayDeserializer
@@ -69,6 +69,10 @@ class KafkaConsole(config: KafkaConfig) {
         Admin.create(props)
     }
 
+    protected def withTimeoutMs[T <: AbstractOptions[T]](options: T) = {
+        options.timeoutMs(timeoutMs)
+    }
+
     private def createAdminClient(): Admin = {
         Admin.create(getProps())
     }

src/main/scala/kafka/console/OperationConsole.scala

@@ -4,11 +4,12 @@ import java.util.concurrent.TimeUnit
 import java.util.{Collections, Properties}
 
 import com.xuxd.kafka.console.config.KafkaConfig
+import org.apache.kafka.clients.admin.ElectLeadersOptions
 import org.apache.kafka.clients.consumer.KafkaConsumer
-import org.apache.kafka.common.TopicPartition
 import org.apache.kafka.common.serialization.ByteArrayDeserializer
+import org.apache.kafka.common.{ElectionType, TopicPartition}
 
-import scala.jdk.CollectionConverters.{CollectionHasAsScala, ListHasAsScala, MapHasAsScala, SeqHasAsJava, SetHasAsScala}
+import scala.jdk.CollectionConverters.{CollectionHasAsScala, ListHasAsScala, MapHasAsScala, SeqHasAsJava, SetHasAsJava, SetHasAsScala}
 
 /**
  * kafka-console-ui.
@@ -194,4 +195,19 @@ class OperationConsole(config: KafkaConfig, topicConsole: TopicConsole,
             thatConsumer.close()
         }
     }
+
+    def electPreferredLeader(partitions: util.Set[TopicPartition]): (Boolean, String) = {
+        withAdminClientAndCatchError(admin => {
+            admin.electLeaders(ElectionType.PREFERRED, partitions, withTimeoutMs(new ElectLeadersOptions)).all().get()
+            (true, "")
+        }, e => {
+            log.error("alter config error.", e)
+            (false, e.getMessage)
+        }).asInstanceOf[(Boolean, String)]
+    }
+
+    def getTopicPartitions(topic: String): util.Set[TopicPartition] = {
+        val topicList = topicConsole.getTopicList(Collections.singleton(topic))
+        topicList.asScala.flatMap(_.partitions().asScala.map(t => new TopicPartition(topic, t.partition()))).toSet.asJava
+    }
 }