Kafka集群动态添加代理

要实现Kafka集群动态添加代理的功能，可以使用Kafka提供的AdminClient API来完成。下面是一个实现该功能的代码示例：

首先，创建一个Java类，命名为"DynamicBrokerAddition.java"，并添加以下代码：

import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.NewPartitions;
import org.apache.kafka.clients.admin.NewPartitionsIncreaseIsrOptions;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.config.TopicConfig;
import org.apache.kafka.common.errors.UnknownTopicOrPartitionException;

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

public class DynamicBrokerAddition {

    private static final String BOOTSTRAP_SERVERS = "localhost:9092"; // Kafka集群的地址

    public static void main(String[] args) {
        String topic = "test-topic"; // 要添加分区的主题
        int numPartitionsToAdd = 1; // 要添加的分区数

        Properties properties = new Properties();
        properties.setProperty(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);

        try (AdminClient adminClient = AdminClient.create(properties)) {
            // 检查主题是否存在
            if (!topicExists(adminClient, topic)) {
                // 创建新的主题
                createTopic(adminClient, topic);
            }

            // 增加分区
            addPartitions(adminClient, topic, numPartitionsToAdd);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static boolean topicExists(AdminClient adminClient, String topic) {
        try {
            adminClient.describeTopics(Collections.singleton(topic)).values().get(topic).get();
            return true;
        } catch (ExecutionException | InterruptedException e) {
            if (e.getCause() instanceof UnknownTopicOrPartitionException) {
                return false;
            }
            throw new RuntimeException(e);
        }
    }

    private static void createTopic(AdminClient adminClient, String topic) {
        NewTopic newTopic = new NewTopic(topic, 1, (short) 1);
        newTopic.configs(Collections.singletonMap(TopicConfig.MIN_IN_SYNC_REPLICAS_CONFIG, "1"));

        try {
            adminClient.createTopics(Collections.singleton(newTopic)).all().get();
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
        }
    }

    private static void addPartitions(AdminClient adminClient, String topic, int numPartitionsToAdd) {
        try {
            NewPartitions newPartitions = NewPartitions.increaseTo(numPartitionsToAdd);
            NewPartitionsIncreaseIsrOptions options = new NewPartitionsIncreaseIsrOptions();
            adminClient.createPartitions(Collections.singletonMap(topic, newPartitions), options).all().get();
        } catch (InterruptedException | ExecutionException e) {
            throw new RuntimeException(e);
        }
    }
}

上述代码中，首先使用AdminClient API检查要添加分区的主题是否存在，如果不存在，则使用AdminClient API创建新的主题。然后，使用AdminClient API增加分区。

要运行上述代码，需要在Java项目中添加以下依赖：

<dependency>
   <groupId>org.apache.kafka</groupId>
   <artifactId>kafka-clients</artifactId>
   <version>2.8.0</version>
</dependency>

请根据实际情况修改代码中的配置，例如Kafka集群地址、主题名称和要添加的分区数。