Kafka中的不公平领导者选举 - 所有分区的相同领导者

在Kafka中，领导者选举是通过ZooKeeper来协调的。默认情况下，Kafka使用的是“公平”领导者选举算法，即每个分区的leader选举独立进行，不同分区的leader可以不一样。但有时候，我们可能希望所有分区的leader都是相同的，这可以通过修改Kafka的配置来实现。

首先，我们需要修改Kafka的配置文件（通常是server.properties），找到以下两个配置项并修改为相同的值：

auto.leader.rebalance.enable=false
unclean.leader.election.enable=false

将这两个配置项的值都设置为false，表示禁用自动的leader rebalance和非安全的leader选举。

然后，我们可以使用Kafka提供的AdminClient来手动指定每个分区的leader。以下是一个示例代码：

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.AlterPartitionReassignmentsOptions;
import org.apache.kafka.clients.admin.AlterPartitionReassignmentsResult;
import org.apache.kafka.clients.admin.NewPartitionReassignment;
import org.apache.kafka.clients.admin.TopicDescription;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.TopicPartition;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class KafkaLeaderElection {
    public static void main(String[] args) {
        // Kafka broker的地址
        String bootstrapServers = "localhost:9092";

        // 创建AdminClient的配置
        Properties config = new Properties();
        config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);

        // 创建AdminClient
        try (AdminClient adminClient = AdminClient.create(config)) {

            // 获取topic的描述信息
            String topicName = "my-topic";
            KafkaFuture<TopicDescription> topicDescriptionFuture = adminClient.describeTopics(Collections.singleton(topicName)).values().get(topicName);
            TopicDescription topicDescription = topicDescriptionFuture.get();

            // 构建leader选举的配置
            Map<TopicPartition, NewPartitionReassignment> reassignments = new HashMap<>();
            for (TopicPartition partition : topicDescription.partitions()) {
                NewPartitionReassignment reassignment = new NewPartitionReassignment(Collections.singletonList(0)); // 设置leader的broker id
                reassignments.put(partition, reassignment);
            }

            // 执行leader选举
            AlterPartitionReassignmentsOptions options = new AlterPartitionReassignmentsOptions().timeoutMs(10000);
            AlterPartitionReassignmentsResult result = adminClient.alterPartitionReassignments(reassignments, options);
            result.all().get();

            System.out.println("Leader election completed successfully!");

        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }
}

在上述示例代码中，我们使用AdminClient来获取指定topic的描述信息，然后构建一个Map来指定每个分区的leader，最后使用alterPartitionReassignments方法来执行leader选举。请注意，这个操作可能需要一些时间来完成，我们可以通过调整timeoutMs来设置超时时间。

通过以上的代码示例和配置修改，我们就可以实现Kafka中所有分区的相同领导者选举了。