第297集消息队列Kafka集群搭建架构实战：高可用部署、性能调优与企业级消息中间件解决方案

/**
 * Kafka集群核心组件
 */
@Component
public class KafkaClusterManager {

    @Autowired
    private BrokerManager brokerManager;

    @Autowired
    private TopicManager topicManager;

    @Autowired
    private ConsumerGroupManager consumerGroupManager;

    @Autowired
    private ClusterMonitorService clusterMonitorService;

    @Autowired
    private SecurityManager securityManager;

    /**
     * 初始化Kafka集群
     */
    public void initializeCluster() {
        try {
            // 1. 初始化Zookeeper连接
            initializeZookeeperConnection();

            // 2. 启动Broker节点
            startBrokerNodes();

            // 3. 创建系统Topic
            createSystemTopics();

            // 4. 配置安全认证
            configureSecurity();

            // 5. 启动监控服务
            startMonitoringService();

            log.info("Kafka集群初始化完成");

        } catch (Exception e) {
            log.error("Kafka集群初始化失败", e);
            throw new ClusterInitializationException("集群初始化失败", e);
        }
    }

    /**
     * 启动Kafka集群
     */
    public void startCluster() {
        try {
            // 1. 检查集群状态
            checkClusterStatus();

            // 2. 启动所有Broker
            brokerManager.startAllBrokers();

            // 3. 验证集群健康状态
            validateClusterHealth();

            // 4. 启动监控
            clusterMonitorService.startMonitoring();

            log.info("Kafka集群启动成功");

        } catch (Exception e) {
            log.error("Kafka集群启动失败", e);
            throw new ClusterStartException("集群启动失败", e);
        }
    }

    /**
     * 停止Kafka集群
     */
    public void stopCluster() {
        try {
            // 1. 停止监控
            clusterMonitorService.stopMonitoring();

            // 2. 停止所有Broker
            brokerManager.stopAllBrokers();

            // 3. 关闭Zookeeper连接
            closeZookeeperConnection();

            log.info("Kafka集群停止成功");

        } catch (Exception e) {
            log.error("Kafka集群停止失败", e);
        }
    }

    /**
     * 初始化Zookeeper连接
     */
    private void initializeZookeeperConnection() {
        // 实现Zookeeper连接初始化逻辑
        log.info("初始化Zookeeper连接");
    }

    /**
     * 启动Broker节点
     */
    private void startBrokerNodes() {
        // 实现Broker节点启动逻辑
        log.info("启动Broker节点");
    }

    /**
     * 创建系统Topic
     */
    private void createSystemTopics() {
        // 实现系统Topic创建逻辑
        log.info("创建系统Topic");
    }

    /**
     * 配置安全认证
     */
    private void configureSecurity() {
        // 实现安全认证配置逻辑
        log.info("配置安全认证");
    }

    /**
     * 启动监控服务
     */
    private void startMonitoringService() {
        // 实现监控服务启动逻辑
        log.info("启动监控服务");
    }
}

/**
 * Kafka Broker节点管理器
 */
@Service
public class BrokerManager {

    @Autowired
    private BrokerConfigService brokerConfigService;

    @Autowired
    private BrokerDeploymentService brokerDeploymentService;

    @Autowired
    private BrokerHealthChecker brokerHealthChecker;

    private final Map<String, BrokerNode> brokerNodes;

    public BrokerManager() {
        this.brokerNodes = new ConcurrentHashMap<>();
    }

    /**
     * 启动所有Broker
     */
    public void startAllBrokers() {
        try {
            // 1. 获取Broker配置
            List<BrokerConfig> brokerConfigs = brokerConfigService.getAllBrokerConfigs();

            // 2. 启动每个Broker
            for (BrokerConfig config : brokerConfigs) {
                startBroker(config);
            }

            // 3. 等待所有Broker启动完成
            waitForBrokersToStart();

            // 4. 验证集群状态
            validateClusterState();

            log.info("所有Broker启动完成，数量: {}", brokerConfigs.size());

        } catch (Exception e) {
            log.error("Broker启动失败", e);
            throw new BrokerStartException("Broker启动失败", e);
        }
    }

    /**
     * 停止所有Broker
     */
    public void stopAllBrokers() {
        try {
            // 1. 优雅关闭所有Broker
            for (BrokerNode brokerNode : brokerNodes.values()) {
                stopBroker(brokerNode);
            }

            // 2. 等待所有Broker停止完成
            waitForBrokersToStop();

            log.info("所有Broker停止完成");

        } catch (Exception e) {
            log.error("Broker停止失败", e);
        }
    }

    /**
     * 启动单个Broker
     */
    private void startBroker(BrokerConfig config) {
        try {
            // 1. 创建Broker节点
            BrokerNode brokerNode = createBrokerNode(config);

            // 2. 部署Broker
            brokerDeploymentService.deployBroker(brokerNode);

            // 3. 启动Broker进程
            brokerDeploymentService.startBrokerProcess(brokerNode);

            // 4. 等待Broker启动
            waitForBrokerToStart(brokerNode);

            // 5. 验证Broker健康状态
            validateBrokerHealth(brokerNode);

            // 6. 注册Broker节点
            brokerNodes.put(brokerNode.getBrokerId(), brokerNode);

            log.info("Broker启动成功: {}", brokerNode.getBrokerId());

        } catch (Exception e) {
            log.error("Broker启动失败: {}", config.getBrokerId(), e);
            throw new BrokerStartException("Broker启动失败", e);
        }
    }

    /**
     * 停止单个Broker
     */
    private void stopBroker(BrokerNode brokerNode) {
        try {
            // 1. 优雅关闭Broker进程
            brokerDeploymentService.stopBrokerProcess(brokerNode);

            // 2. 等待Broker停止
            waitForBrokerToStop(brokerNode);

            // 3. 清理Broker资源
            brokerDeploymentService.cleanupBrokerResources(brokerNode);

            // 4. 注销Broker节点
            brokerNodes.remove(brokerNode.getBrokerId());

            log.info("Broker停止成功: {}", brokerNode.getBrokerId());

        } catch (Exception e) {
            log.error("Broker停止失败: {}", brokerNode.getBrokerId(), e);
        }
    }

    /**
     * 创建Broker节点
     */
    private BrokerNode createBrokerNode(BrokerConfig config) {
        BrokerNode brokerNode = new BrokerNode();
        brokerNode.setBrokerId(config.getBrokerId());
        brokerNode.setHost(config.getHost());
        brokerNode.setPort(config.getPort());
        brokerNode.setLogDir(config.getLogDir());
        brokerNode.setConfig(config);
        brokerNode.setStatus(BrokerStatus.STARTING);
        brokerNode.setStartTime(System.currentTimeMillis());

        return brokerNode;
    }

    /**
     * 等待Broker启动
     */
    private void waitForBrokerToStart(BrokerNode brokerNode) {
        int maxWaitTime = 30000; // 30秒
        int checkInterval = 1000; // 1秒
        int waitedTime = 0;

        while (waitedTime < maxWaitTime) {
            try {
                if (brokerHealthChecker.isBrokerHealthy(brokerNode)) {
                    brokerNode.setStatus(BrokerStatus.RUNNING);
                    return;
                }

                Thread.sleep(checkInterval);
                waitedTime += checkInterval;

            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new BrokerStartException("等待Broker启动被中断", e);
            }
        }

        throw new BrokerStartException("Broker启动超时: " + brokerNode.getBrokerId());
    }

    /**
     * 等待Broker停止
     */
    private void waitForBrokerToStop(BrokerNode brokerNode) {
        int maxWaitTime = 30000; // 30秒
        int checkInterval = 1000; // 1秒
        int waitedTime = 0;

        while (waitedTime < maxWaitTime) {
            try {
                if (!brokerHealthChecker.isBrokerRunning(brokerNode)) {
                    brokerNode.setStatus(BrokerStatus.STOPPED);
                    return;
                }

                Thread.sleep(checkInterval);
                waitedTime += checkInterval;

            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new BrokerStopException("等待Broker停止被中断", e);
            }
        }

        throw new BrokerStopException("Broker停止超时: " + brokerNode.getBrokerId());
    }

    /**
     * 验证Broker健康状态
     */
    private void validateBrokerHealth(BrokerNode brokerNode) {
        try {
            BrokerHealthStatus healthStatus = brokerHealthChecker.checkBrokerHealth(brokerNode);

            if (!healthStatus.isHealthy()) {
                throw new BrokerHealthException("Broker健康检查失败: " + brokerNode.getBrokerId());
            }

            brokerNode.setHealthStatus(healthStatus);

        } catch (Exception e) {
            log.error("Broker健康检查失败: {}", brokerNode.getBrokerId(), e);
            throw new BrokerHealthException("Broker健康检查失败", e);
        }
    }

    /**
     * 等待所有Broker启动完成
     */
    private void waitForBrokersToStart() {
        int maxWaitTime = 120000; // 2分钟
        int checkInterval = 5000; // 5秒
        int waitedTime = 0;

        while (waitedTime < maxWaitTime) {
            try {
                boolean allBrokersRunning = brokerNodes.values().stream()
                    .allMatch(broker -> broker.getStatus() == BrokerStatus.RUNNING);

                if (allBrokersRunning) {
                    return;
                }

                Thread.sleep(checkInterval);
                waitedTime += checkInterval;

            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new BrokerStartException("等待所有Broker启动被中断", e);
            }
        }

        throw new BrokerStartException("等待所有Broker启动超时");
    }

    /**
     * 等待所有Broker停止完成
     */
    private void waitForBrokersToStop() {
        int maxWaitTime = 120000; // 2分钟
        int checkInterval = 5000; // 5秒
        int waitedTime = 0;

        while (waitedTime < maxWaitTime) {
            try {
                boolean allBrokersStopped = brokerNodes.values().stream()
                    .allMatch(broker -> broker.getStatus() == BrokerStatus.STOPPED);

                if (allBrokersStopped) {
                    return;
                }

                Thread.sleep(checkInterval);
                waitedTime += checkInterval;

            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new BrokerStopException("等待所有Broker停止被中断", e);
            }
        }

        throw new BrokerStopException("等待所有Broker停止超时");
    }

    /**
     * 验证集群状态
     */
    private void validateClusterState() {
        try {
            // 检查集群是否形成
            if (brokerNodes.size() < 3) {
                log.warn("集群节点数量不足，建议至少3个节点");
            }

            // 检查所有Broker是否健康
            for (BrokerNode brokerNode : brokerNodes.values()) {
                if (brokerNode.getHealthStatus() == null || !brokerNode.getHealthStatus().isHealthy()) {
                    throw new ClusterValidationException("集群状态验证失败，Broker不健康: " + brokerNode.getBrokerId());
                }
            }

            log.info("集群状态验证通过");

        } catch (Exception e) {
            log.error("集群状态验证失败", e);
            throw new ClusterValidationException("集群状态验证失败", e);
        }
    }
}

/**
 * Kafka Broker配置服务
 */
@Service
public class BrokerConfigService {

    @Autowired
    private ConfigurationRepository configRepository;

    /**
     * 获取所有Broker配置
     */
    public List<BrokerConfig> getAllBrokerConfigs() {
        return configRepository.findAllBrokerConfigs();
    }

    /**
     * 获取Broker配置
     */
    public BrokerConfig getBrokerConfig(String brokerId) {
        return configRepository.findBrokerConfigById(brokerId)
            .orElseThrow(() -> new ConfigNotFoundException("Broker配置不存在: " + brokerId));
    }

    /**
     * 保存Broker配置
     */
    public void saveBrokerConfig(BrokerConfig config) {
        try {
            // 验证配置
            validateBrokerConfig(config);

            // 保存配置
            configRepository.saveBrokerConfig(config);

            log.info("Broker配置保存成功: {}", config.getBrokerId());

        } catch (Exception e) {
            log.error("Broker配置保存失败", e);
            throw new ConfigSaveException("Broker配置保存失败", e);
        }
    }

    /**
     * 更新Broker配置
     */
    public void updateBrokerConfig(String brokerId, BrokerConfig config) {
        try {
            // 检查配置是否存在
            if (!configRepository.existsBrokerConfig(brokerId)) {
                throw new ConfigNotFoundException("Broker配置不存在: " + brokerId);
            }

            // 验证配置
            validateBrokerConfig(config);

            // 更新配置
            config.setBrokerId(brokerId);
            configRepository.saveBrokerConfig(config);

            log.info("Broker配置更新成功: {}", brokerId);

        } catch (Exception e) {
            log.error("Broker配置更新失败", e);
            throw new ConfigUpdateException("Broker配置更新失败", e);
        }
    }

    /**
     * 删除Broker配置
     */
    public void deleteBrokerConfig(String brokerId) {
        try {
            if (!configRepository.existsBrokerConfig(brokerId)) {
                throw new ConfigNotFoundException("Broker配置不存在: " + brokerId);
            }

            configRepository.deleteBrokerConfig(brokerId);

            log.info("Broker配置删除成功: {}", brokerId);

        } catch (Exception e) {
            log.error("Broker配置删除失败", e);
            throw new ConfigDeleteException("Broker配置删除失败", e);
        }
    }

    /**
     * 生成Broker配置文件
     */
    public String generateBrokerConfigFile(BrokerConfig config) {
        StringBuilder configContent = new StringBuilder();

        // 基础配置
        configContent.append("# Kafka Broker Configuration\n");
        configContent.append("broker.id=").append(config.getBrokerId()).append("\n");
        configContent.append("listeners=PLAINTEXT://").append(config.getHost()).append(":").append(config.getPort()).append("\n");
        configContent.append("advertised.listeners=PLAINTEXT://").append(config.getHost()).append(":").append(config.getPort()).append("\n");

        // 日志配置
        configContent.append("log.dirs=").append(config.getLogDir()).append("\n");
        configContent.append("num.network.threads=").append(config.getNumNetworkThreads()).append("\n");
        configContent.append("num.io.threads=").append(config.getNumIoThreads()).append("\n");
        configContent.append("socket.send.buffer.bytes=").append(config.getSocketSendBufferBytes()).append("\n");
        configContent.append("socket.receive.buffer.bytes=").append(config.getSocketReceiveBufferBytes()).append("\n");
        configContent.append("socket.request.max.bytes=").append(config.getSocketRequestMaxBytes()).append("\n");

        // 日志保留配置
        configContent.append("log.retention.hours=").append(config.getLogRetentionHours()).append("\n");
        configContent.append("log.retention.bytes=").append(config.getLogRetentionBytes()).append("\n");
        configContent.append("log.segment.bytes=").append(config.getLogSegmentBytes()).append("\n");
        configContent.append("log.retention.check.interval.ms=").append(config.getLogRetentionCheckIntervalMs()).append("\n");

        // 副本配置
        configContent.append("default.replication.factor=").append(config.getDefaultReplicationFactor()).append("\n");
        configContent.append("min.insync.replicas=").append(config.getMinInsyncReplicas()).append("\n");

        // Zookeeper配置
        configContent.append("zookeeper.connect=").append(config.getZookeeperConnect()).append("\n");
        configContent.append("zookeeper.connection.timeout.ms=").append(config.getZookeeperConnectionTimeoutMs()).append("\n");

        // 性能配置
        configContent.append("num.partitions=").append(config.getNumPartitions()).append("\n");
        configContent.append("offsets.topic.replication.factor=").append(config.getOffsetsTopicReplicationFactor()).append("\n");
        configContent.append("transaction.state.log.replication.factor=").append(config.getTransactionStateLogReplicationFactor()).append("\n");
        configContent.append("transaction.state.log.min.isr=").append(config.getTransactionStateLogMinIsr()).append("\n");

        // 安全配置
        if (config.isSecurityEnabled()) {
            configContent.append("security.inter.broker.protocol=").append(config.getSecurityInterBrokerProtocol()).append("\n");
            configContent.append("sasl.mechanism.inter.broker.protocol=").append(config.getSaslMechanismInterBrokerProtocol()).append("\n");
            configContent.append("sasl.enabled.mechanisms=").append(config.getSaslEnabledMechanisms()).append("\n");
        }

        return configContent.toString();
    }

    /**
     * 验证Broker配置
     */
    private void validateBrokerConfig(BrokerConfig config) {
        if (config.getBrokerId() == null || config.getBrokerId().isEmpty()) {
            throw new ConfigValidationException("Broker ID不能为空");
        }

        if (config.getHost() == null || config.getHost().isEmpty()) {
            throw new ConfigValidationException("Broker主机地址不能为空");
        }

        if (config.getPort() <= 0 || config.getPort() > 65535) {
            throw new ConfigValidationException("Broker端口必须在1-65535之间");
        }

        if (config.getLogDir() == null || config.getLogDir().isEmpty()) {
            throw new ConfigValidationException("日志目录不能为空");
        }

        if (config.getZookeeperConnect() == null || config.getZookeeperConnect().isEmpty()) {
            throw new ConfigValidationException("Zookeeper连接地址不能为空");
        }

        if (config.getDefaultReplicationFactor() < 1) {
            throw new ConfigValidationException("默认副本因子必须大于0");
        }

        if (config.getMinInsyncReplicas() < 1) {
            throw new ConfigValidationException("最小同步副本数必须大于0");
        }

        if (config.getMinInsyncReplicas() > config.getDefaultReplicationFactor()) {
            throw new ConfigValidationException("最小同步副本数不能大于默认副本因子");
        }
    }
}

/**
 * Kafka Broker部署服务
 */
@Service
public class BrokerDeploymentService {

    @Autowired
    private FileSystemService fileSystemService;

    @Autowired
    private ProcessManager processManager;

    @Autowired
    private NetworkService networkService;

    /**
     * 部署Broker
     */
    public void deployBroker(BrokerNode brokerNode) {
        try {
            // 1. 创建Broker目录
            createBrokerDirectories(brokerNode);

            // 2. 生成配置文件
            generateBrokerConfigFile(brokerNode);

            // 3. 创建启动脚本
            createBrokerStartScript(brokerNode);

            // 4. 创建停止脚本
            createBrokerStopScript(brokerNode);

            // 5. 设置文件权限
            setBrokerFilePermissions(brokerNode);

            log.info("Broker部署完成: {}", brokerNode.getBrokerId());

        } catch (Exception e) {
            log.error("Broker部署失败: {}", brokerNode.getBrokerId(), e);
            throw new BrokerDeploymentException("Broker部署失败", e);
        }
    }

    /**
     * 启动Broker进程
     */
    public void startBrokerProcess(BrokerNode brokerNode) {
        try {
            // 1. 检查端口是否可用
            if (!networkService.isPortAvailable(brokerNode.getHost(), brokerNode.getPort())) {
                throw new BrokerStartException("端口不可用: " + brokerNode.getPort());
            }

            // 2. 启动Broker进程
            Process process = processManager.startProcess(
                brokerNode.getStartScriptPath(),
                brokerNode.getConfigFilePath(),
                brokerNode.getLogFilePath()
            );

            // 3. 设置进程信息
            brokerNode.setProcess(process);
            brokerNode.setPid(process.pid());

            log.info("Broker进程启动成功: {}, PID: {}", brokerNode.getBrokerId(), process.pid());

        } catch (Exception e) {
            log.error("Broker进程启动失败: {}", brokerNode.getBrokerId(), e);
            throw new BrokerStartException("Broker进程启动失败", e);
        }
    }

    /**
     * 停止Broker进程
     */
    public void stopBrokerProcess(BrokerNode brokerNode) {
        try {
            if (brokerNode.getProcess() != null && brokerNode.getProcess().isAlive()) {
                // 1. 优雅关闭进程
                brokerNode.getProcess().destroy();

                // 2. 等待进程结束
                boolean terminated = brokerNode.getProcess().waitFor(30, TimeUnit.SECONDS);

                if (!terminated) {
                    // 3. 强制关闭进程
                    brokerNode.getProcess().destroyForcibly();
                    brokerNode.getProcess().waitFor(10, TimeUnit.SECONDS);
                }

                log.info("Broker进程停止成功: {}", brokerNode.getBrokerId());
            }

        } catch (Exception e) {
            log.error("Broker进程停止失败: {}", brokerNode.getBrokerId(), e);
            throw new BrokerStopException("Broker进程停止失败", e);
        }
    }

    /**
     * 清理Broker资源
     */
    public void cleanupBrokerResources(BrokerNode brokerNode) {
        try {
            // 1. 清理临时文件
            fileSystemService.cleanupTempFiles(brokerNode.getTempDir());

            // 2. 清理日志文件
            fileSystemService.cleanupLogFiles(brokerNode.getLogDir());

            // 3. 清理配置文件
            fileSystemService.cleanupConfigFiles(brokerNode.getConfigDir());

            log.info("Broker资源清理完成: {}", brokerNode.getBrokerId());

        } catch (Exception e) {
            log.error("Broker资源清理失败: {}", brokerNode.getBrokerId(), e);
        }
    }

    /**
     * 创建Broker目录
     */
    private void createBrokerDirectories(BrokerNode brokerNode) {
        try {
            // 创建基础目录
            fileSystemService.createDirectory(brokerNode.getBaseDir());
            fileSystemService.createDirectory(brokerNode.getConfigDir());
            fileSystemService.createDirectory(brokerNode.getLogDir());
            fileSystemService.createDirectory(brokerNode.getTempDir());
            fileSystemService.createDirectory(brokerNode.getScriptDir());

            // 创建日志子目录
            fileSystemService.createDirectory(brokerNode.getLogDir() + "/kafka-logs");
            fileSystemService.createDirectory(brokerNode.getLogDir() + "/kafka-logs-0");
            fileSystemService.createDirectory(brokerNode.getLogDir() + "/kafka-logs-1");
            fileSystemService.createDirectory(brokerNode.getLogDir() + "/kafka-logs-2");

        } catch (Exception e) {
            log.error("创建Broker目录失败: {}", brokerNode.getBrokerId(), e);
            throw new BrokerDeploymentException("创建Broker目录失败", e);
        }
    }

    /**
     * 生成Broker配置文件
     */
    private void generateBrokerConfigFile(BrokerNode brokerNode) {
        try {
            BrokerConfig config = brokerNode.getConfig();
            String configContent = generateBrokerConfigContent(config);

            String configFilePath = brokerNode.getConfigDir() + "/server.properties";
            fileSystemService.writeFile(configFilePath, configContent);

            brokerNode.setConfigFilePath(configFilePath);

        } catch (Exception e) {
            log.error("生成Broker配置文件失败: {}", brokerNode.getBrokerId(), e);
            throw new BrokerDeploymentException("生成Broker配置文件失败", e);
        }
    }

    /**
     * 创建Broker启动脚本
     */
    private void createBrokerStartScript(BrokerNode brokerNode) {
        try {
            String startScriptContent = generateStartScriptContent(brokerNode);

            String startScriptPath = brokerNode.getScriptDir() + "/start-kafka.sh";
            fileSystemService.writeFile(startScriptPath, startScriptContent);

            brokerNode.setStartScriptPath(startScriptPath);

        } catch (Exception e) {
            log.error("创建Broker启动脚本失败: {}", brokerNode.getBrokerId(), e);
            throw new BrokerDeploymentException("创建Broker启动脚本失败", e);
        }
    }

    /**
     * 创建Broker停止脚本
     */
    private void createBrokerStopScript(BrokerNode brokerNode) {
        try {
            String stopScriptContent = generateStopScriptContent(brokerNode);

            String stopScriptPath = brokerNode.getScriptDir() + "/stop-kafka.sh";
            fileSystemService.writeFile(stopScriptPath, stopScriptContent);

            brokerNode.setStopScriptPath(stopScriptPath);

        } catch (Exception e) {
            log.error("创建Broker停止脚本失败: {}", brokerNode.getBrokerId(), e);
            throw new BrokerDeploymentException("创建Broker停止脚本失败", e);
        }
    }

    /**
     * 设置Broker文件权限
     */
    private void setBrokerFilePermissions(BrokerNode brokerNode) {
        try {
            // 设置脚本文件执行权限
            fileSystemService.setExecutablePermission(brokerNode.getStartScriptPath());
            fileSystemService.setExecutablePermission(brokerNode.getStopScriptPath());

            // 设置配置文件读取权限
            fileSystemService.setReadablePermission(brokerNode.getConfigFilePath());

        } catch (Exception e) {
            log.error("设置Broker文件权限失败: {}", brokerNode.getBrokerId(), e);
            throw new BrokerDeploymentException("设置Broker文件权限失败", e);
        }
    }

    /**
     * 生成Broker配置内容
     */
    private String generateBrokerConfigContent(BrokerConfig config) {
        StringBuilder content = new StringBuilder();

        content.append("# Kafka Broker Configuration\n");
        content.append("broker.id=").append(config.getBrokerId()).append("\n");
        content.append("listeners=PLAINTEXT://").append(config.getHost()).append(":").append(config.getPort()).append("\n");
        content.append("advertised.listeners=PLAINTEXT://").append(config.getHost()).append(":").append(config.getPort()).append("\n");
        content.append("log.dirs=").append(config.getLogDir()).append("\n");
        content.append("num.network.threads=").append(config.getNumNetworkThreads()).append("\n");
        content.append("num.io.threads=").append(config.getNumIoThreads()).append("\n");
        content.append("socket.send.buffer.bytes=").append(config.getSocketSendBufferBytes()).append("\n");
        content.append("socket.receive.buffer.bytes=").append(config.getSocketReceiveBufferBytes()).append("\n");
        content.append("socket.request.max.bytes=").append(config.getSocketRequestMaxBytes()).append("\n");
        content.append("log.retention.hours=").append(config.getLogRetentionHours()).append("\n");
        content.append("log.retention.bytes=").append(config.getLogRetentionBytes()).append("\n");
        content.append("log.segment.bytes=").append(config.getLogSegmentBytes()).append("\n");
        content.append("log.retention.check.interval.ms=").append(config.getLogRetentionCheckIntervalMs()).append("\n");
        content.append("default.replication.factor=").append(config.getDefaultReplicationFactor()).append("\n");
        content.append("min.insync.replicas=").append(config.getMinInsyncReplicas()).append("\n");
        content.append("zookeeper.connect=").append(config.getZookeeperConnect()).append("\n");
        content.append("zookeeper.connection.timeout.ms=").append(config.getZookeeperConnectionTimeoutMs()).append("\n");
        content.append("num.partitions=").append(config.getNumPartitions()).append("\n");
        content.append("offsets.topic.replication.factor=").append(config.getOffsetsTopicReplicationFactor()).append("\n");
        content.append("transaction.state.log.replication.factor=").append(config.getTransactionStateLogReplicationFactor()).append("\n");
        content.append("transaction.state.log.min.isr=").append(config.getTransactionStateLogMinIsr()).append("\n");

        return content.toString();
    }

    /**
     * 生成启动脚本内容
     */
    private String generateStartScriptContent(BrokerNode brokerNode) {
        StringBuilder content = new StringBuilder();

        content.append("#!/bin/bash\n");
        content.append("# Kafka Broker Start Script\n");
        content.append("export KAFKA_HOME=").append(brokerNode.getKafkaHome()).append("\n");
        content.append("export KAFKA_LOG_DIR=").append(brokerNode.getLogDir()).append("\n");
        content.append("export KAFKA_CONFIG_DIR=").append(brokerNode.getConfigDir()).append("\n");
        content.append("\n");
        content.append("echo \"Starting Kafka Broker ").append(brokerNode.getBrokerId()).append("...\"\n");
        content.append("nohup $KAFKA_HOME/bin/kafka-server-start.sh $KAFKA_CONFIG_DIR/server.properties > $KAFKA_LOG_DIR/kafka.out 2>&1 &\n");
        content.append("echo \"Kafka Broker ").append(brokerNode.getBrokerId()).append(" started\"\n");

        return content.toString();
    }

    /**
     * 生成停止脚本内容
     */
    private String generateStopScriptContent(BrokerNode brokerNode) {
        StringBuilder content = new StringBuilder();

        content.append("#!/bin/bash\n");
        content.append("# Kafka Broker Stop Script\n");
        content.append("export KAFKA_HOME=").append(brokerNode.getKafkaHome()).append("\n");
        content.append("\n");
        content.append("echo \"Stopping Kafka Broker ").append(brokerNode.getBrokerId()).append("...\"\n");
        content.append("$KAFKA_HOME/bin/kafka-server-stop.sh\n");
        content.append("echo \"Kafka Broker ").append(brokerNode.getBrokerId()).append(" stopped\"\n");

        return content.toString();
    }
}

/**
 * Kafka Topic管理器
 */
@Service
public class TopicManager {

    @Autowired
    private KafkaAdminClient kafkaAdminClient;

    @Autowired
    private TopicConfigService topicConfigService;

    @Autowired
    private PartitionManager partitionManager;

    /**
     * 创建Topic
     */
    public void createTopic(TopicConfig topicConfig) {
        try {
            // 1. 验证Topic配置
            validateTopicConfig(topicConfig);

            // 2. 检查Topic是否已存在
            if (topicExists(topicConfig.getTopicName())) {
                throw new TopicExistsException("Topic已存在: " + topicConfig.getTopicName());
            }

            // 3. 创建Topic
            NewTopic newTopic = new NewTopic(
                topicConfig.getTopicName(),
                topicConfig.getNumPartitions(),
                topicConfig.getReplicationFactor()
            );

            // 4. 设置Topic配置
            Map<String, String> configs = topicConfig.getConfigs();
            if (configs != null && !configs.isEmpty()) {
                newTopic.configs(configs);
            }

            // 5. 执行创建
            CreateTopicsResult result = kafkaAdminClient.createTopics(Collections.singletonList(newTopic));

            // 6. 等待创建完成
            result.all().get(30, TimeUnit.SECONDS);

            // 7. 保存Topic配置
            topicConfigService.saveTopicConfig(topicConfig);

            log.info("Topic创建成功: {}", topicConfig.getTopicName());

        } catch (Exception e) {
            log.error("Topic创建失败: {}", topicConfig.getTopicName(), e);
            throw new TopicCreationException("Topic创建失败", e);
        }
    }

    /**
     * 删除Topic
     */
    public void deleteTopic(String topicName) {
        try {
            // 1. 检查Topic是否存在
            if (!topicExists(topicName)) {
                throw new TopicNotFoundException("Topic不存在: " + topicName);
            }

            // 2. 删除Topic
            DeleteTopicsResult result = kafkaAdminClient.deleteTopics(Collections.singletonList(topicName));

            // 3. 等待删除完成
            result.all().get(30, TimeUnit.SECONDS);

            // 4. 删除Topic配置
            topicConfigService.deleteTopicConfig(topicName);

            log.info("Topic删除成功: {}", topicName);

        } catch (Exception e) {
            log.error("Topic删除失败: {}", topicName, e);
            throw new TopicDeletionException("Topic删除失败", e);
        }
    }

    /**
     * 更新Topic配置
     */
    public void updateTopicConfig(String topicName, Map<String, String> configs) {
        try {
            // 1. 检查Topic是否存在
            if (!topicExists(topicName)) {
                throw new TopicNotFoundException("Topic不存在: " + topicName);
            }

            // 2. 更新Topic配置
            ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, topicName);
            Config config = new Config(configs.entrySet().stream()
                .map(entry -> new ConfigEntry(entry.getKey(), entry.getValue()))
                .collect(Collectors.toList()));

            AlterConfigsResult result = kafkaAdminClient.alterConfigs(
                Collections.singletonMap(configResource, config)
            );

            // 3. 等待更新完成
            result.all().get(30, TimeUnit.SECONDS);

            // 4. 更新Topic配置
            topicConfigService.updateTopicConfig(topicName, configs);

            log.info("Topic配置更新成功: {}", topicName);

        } catch (Exception e) {
            log.error("Topic配置更新失败: {}", topicName, e);
            throw new TopicConfigUpdateException("Topic配置更新失败", e);
        }
    }

    /**
     * 获取Topic信息
     */
    public TopicInfo getTopicInfo(String topicName) {
        try {
            // 1. 检查Topic是否存在
            if (!topicExists(topicName)) {
                throw new TopicNotFoundException("Topic不存在: " + topicName);
            }

            // 2. 获取Topic详细信息
            DescribeTopicsResult result = kafkaAdminClient.describeTopics(Collections.singletonList(topicName));
            TopicDescription topicDescription = result.all().get().get(topicName);

            // 3. 获取Topic配置
            ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, topicName);
            DescribeConfigsResult configResult = kafkaAdminClient.describeConfigs(Collections.singletonList(configResource));
            Config config = configResult.all().get().get(configResource);

            // 4. 构建Topic信息
            TopicInfo topicInfo = new TopicInfo();
            topicInfo.setTopicName(topicName);
            topicInfo.setPartitions(topicDescription.partitions().size());
            topicInfo.setReplicationFactor(topicDescription.partitions().get(0).replicas().size());
            topicInfo.setConfigs(config.entries().stream()
                .collect(Collectors.toMap(ConfigEntry::name, ConfigEntry::value)));

            return topicInfo;

        } catch (Exception e) {
            log.error("获取Topic信息失败: {}", topicName, e);
            throw new TopicInfoException("获取Topic信息失败", e);
        }
    }

    /**
     * 获取所有Topic列表
     */
    public List<TopicInfo> getAllTopics() {
        try {
            // 1. 获取所有Topic名称
            ListTopicsResult result = kafkaAdminClient.listTopics();
            Set<String> topicNames = result.names().get();

            // 2. 获取每个Topic的详细信息
            List<TopicInfo> topicInfos = new ArrayList<>();
            for (String topicName : topicNames) {
                try {
                    TopicInfo topicInfo = getTopicInfo(topicName);
                    topicInfos.add(topicInfo);
                } catch (Exception e) {
                    log.warn("获取Topic信息失败: {}", topicName, e);
                }
            }

            return topicInfos;

        } catch (Exception e) {
            log.error("获取所有Topic列表失败", e);
            throw new TopicListException("获取所有Topic列表失败", e);
        }
    }

    /**
     * 检查Topic是否存在
     */
    public boolean topicExists(String topicName) {
        try {
            ListTopicsResult result = kafkaAdminClient.listTopics();
            Set<String> topicNames = result.names().get();
            return topicNames.contains(topicName);

        } catch (Exception e) {
            log.error("检查Topic是否存在失败: {}", topicName, e);
            return false;
        }
    }

    /**
     * 验证Topic配置
     */
    private void validateTopicConfig(TopicConfig topicConfig) {
        if (topicConfig.getTopicName() == null || topicConfig.getTopicName().isEmpty()) {
            throw new TopicConfigValidationException("Topic名称不能为空");
        }

        if (topicConfig.getNumPartitions() <= 0) {
            throw new TopicConfigValidationException("分区数必须大于0");
        }

        if (topicConfig.getReplicationFactor() <= 0) {
            throw new TopicConfigValidationException("副本因子必须大于0");
        }

        if (topicConfig.getReplicationFactor() > getMaxReplicationFactor()) {
            throw new TopicConfigValidationException("副本因子不能大于最大副本因子");
        }

        // 验证Topic名称格式
        if (!isValidTopicName(topicConfig.getTopicName())) {
            throw new TopicConfigValidationException("Topic名称格式不正确");
        }
    }

    /**
     * 获取最大副本因子
     */
    private int getMaxReplicationFactor() {
        // 实现获取最大副本因子的逻辑
        return 3; // 默认最大副本因子
    }

    /**
     * 验证Topic名称格式
     */
    private boolean isValidTopicName(String topicName) {
        // Topic名称只能包含字母、数字、点、下划线和连字符
        return topicName.matches("^[a-zA-Z0-9._-]+$");
    }
}

/**
 * Kafka分区管理器
 */
@Service
public class PartitionManager {

    @Autowired
    private KafkaAdminClient kafkaAdminClient;

    @Autowired
    private BrokerManager brokerManager;

    /**
     * 增加Topic分区数
     */
    public void increasePartitions(String topicName, int newPartitionCount) {
        try {
            // 1. 检查Topic是否存在
            if (!topicExists(topicName)) {
                throw new TopicNotFoundException("Topic不存在: " + topicName);
            }

            // 2. 获取当前分区数
            int currentPartitionCount = getCurrentPartitionCount(topicName);

            // 3. 验证新分区数
            if (newPartitionCount <= currentPartitionCount) {
                throw new PartitionValidationException("新分区数必须大于当前分区数");
            }

            // 4. 创建分区变更请求
            Map<String, NewPartitions> partitionsMap = new HashMap<>();
            partitionsMap.put(topicName, NewPartitions.increaseTo(newPartitionCount));

            // 5. 执行分区变更
            CreatePartitionsResult result = kafkaAdminClient.createPartitions(partitionsMap);

            // 6. 等待变更完成
            result.all().get(30, TimeUnit.SECONDS);

            log.info("Topic分区数增加成功: {} -> {}", topicName, newPartitionCount);

        } catch (Exception e) {
            log.error("增加Topic分区数失败: {}", topicName, e);
            throw new PartitionIncreaseException("增加Topic分区数失败", e);
        }
    }

    /**
     * 重新分配分区
     */
    public void reassignPartitions(String topicName, Map<Integer, List<Integer>> partitionReplicas) {
        try {
            // 1. 检查Topic是否存在
            if (!topicExists(topicName)) {
                throw new TopicNotFoundException("Topic不存在: " + topicName);
            }

            // 2. 验证分区分配
            validatePartitionReplicas(topicName, partitionReplicas);

            // 3. 创建分区重分配请求
            Map<TopicPartition, Optional<NewPartitionReassignment>> reassignments = new HashMap<>();

            for (Map.Entry<Integer, List<Integer>> entry : partitionReplicas.entrySet()) {
                int partition = entry.getKey();
                List<Integer> replicas = entry.getValue();

                TopicPartition topicPartition = new TopicPartition(topicName, partition);
                NewPartitionReassignment reassignment = new NewPartitionReassignment(replicas);
                reassignments.put(topicPartition, Optional.of(reassignment));
            }

            // 4. 执行分区重分配
            AlterPartitionReassignmentsResult result = kafkaAdminClient.alterPartitionReassignments(reassignments);

            // 5. 等待重分配完成
            result.all().get(60, TimeUnit.SECONDS);

            log.info("Topic分区重分配成功: {}", topicName);

        } catch (Exception e) {
            log.error("分区重分配失败: {}", topicName, e);
            throw new PartitionReassignmentException("分区重分配失败", e);
        }
    }

    /**
     * 获取分区信息
     */
    public List<PartitionInfo> getPartitionInfo(String topicName) {
        try {
            // 1. 检查Topic是否存在
            if (!topicExists(topicName)) {
                throw new TopicNotFoundException("Topic不存在: " + topicName);
            }

            // 2. 获取Topic详细信息
            DescribeTopicsResult result = kafkaAdminClient.describeTopics(Collections.singletonList(topicName));
            TopicDescription topicDescription = result.all().get().get(topicName);

            // 3. 构建分区信息
            List<PartitionInfo> partitionInfos = new ArrayList<>();
            for (TopicPartitionInfo partitionInfo : topicDescription.partitions()) {
                PartitionInfo info = new PartitionInfo();
                info.setTopicName(topicName);
                info.setPartition(partitionInfo.partition());
                info.setLeader(partitionInfo.leader().id());
                info.setReplicas(partitionInfo.replicas().stream()
                    .map(Node::id)
                    .collect(Collectors.toList()));
                info.setIsr(partitionInfo.isr().stream()
                    .map(Node::id)
                    .collect(Collectors.toList()));

                partitionInfos.add(info);
            }

            return partitionInfos;

        } catch (Exception e) {
            log.error("获取分区信息失败: {}", topicName, e);
            throw new PartitionInfoException("获取分区信息失败", e);
        }
    }

    /**
     * 检查分区健康状态
     */
    public PartitionHealthStatus checkPartitionHealth(String topicName) {
        try {
            // 1. 获取分区信息
            List<PartitionInfo> partitionInfos = getPartitionInfo(topicName);

            // 2. 分析分区健康状态
            PartitionHealthStatus healthStatus = new PartitionHealthStatus();
            healthStatus.setTopicName(topicName);
            healthStatus.setTotalPartitions(partitionInfos.size());

            int healthyPartitions = 0;
            int underReplicatedPartitions = 0;
            int leaderlessPartitions = 0;

            for (PartitionInfo partitionInfo : partitionInfos) {
                // 检查副本数
                int expectedReplicas = partitionInfo.getReplicas().size();
                int actualReplicas = partitionInfo.getIsr().size();

                if (actualReplicas < expectedReplicas) {
                    underReplicatedPartitions++;
                }

                // 检查Leader
                if (partitionInfo.getLeader() == -1) {
                    leaderlessPartitions++;
                }

                // 检查分区是否健康
                if (actualReplicas >= expectedReplicas && partitionInfo.getLeader() != -1) {
                    healthyPartitions++;
                }
            }

            healthStatus.setHealthyPartitions(healthyPartitions);
            healthStatus.setUnderReplicatedPartitions(underReplicatedPartitions);
            healthStatus.setLeaderlessPartitions(leaderlessPartitions);

            // 计算健康率
            double healthRate = (double) healthyPartitions / partitionInfos.size();
            healthStatus.setHealthRate(healthRate);

            // 判断整体健康状态
            healthStatus.setHealthy(healthRate >= 0.8); // 80%以上分区健康

            return healthStatus;

        } catch (Exception e) {
            log.error("检查分区健康状态失败: {}", topicName, e);
            throw new PartitionHealthException("检查分区健康状态失败", e);
        }
    }

    /**
     * 获取当前分区数
     */
    private int getCurrentPartitionCount(String topicName) {
        try {
            DescribeTopicsResult result = kafkaAdminClient.describeTopics(Collections.singletonList(topicName));
            TopicDescription topicDescription = result.all().get().get(topicName);
            return topicDescription.partitions().size();

        } catch (Exception e) {
            log.error("获取当前分区数失败: {}", topicName, e);
            throw new PartitionInfoException("获取当前分区数失败", e);
        }
    }

    /**
     * 验证分区副本分配
     */
    private void validatePartitionReplicas(String topicName, Map<Integer, List<Integer>> partitionReplicas) {
        // 1. 检查分区是否存在
        List<PartitionInfo> currentPartitions = getPartitionInfo(topicName);
        Set<Integer> existingPartitions = currentPartitions.stream()
            .map(PartitionInfo::getPartition)
            .collect(Collectors.toSet());

        for (Integer partition : partitionReplicas.keySet()) {
            if (!existingPartitions.contains(partition)) {
                throw new PartitionValidationException("分区不存在: " + partition);
            }
        }

        // 2. 检查副本数
        for (Map.Entry<Integer, List<Integer>> entry : partitionReplicas.entrySet()) {
            List<Integer> replicas = entry.getValue();

            if (replicas.isEmpty()) {
                throw new PartitionValidationException("分区副本不能为空: " + entry.getKey());
            }

            // 检查副本是否重复
            Set<Integer> replicaSet = new HashSet<>(replicas);
            if (replicaSet.size() != replicas.size()) {
                throw new PartitionValidationException("分区副本不能重复: " + entry.getKey());
            }

            // 检查副本是否有效
            for (Integer replica : replicas) {
                if (!brokerManager.isBrokerExists(replica)) {
                    throw new PartitionValidationException("副本Broker不存在: " + replica);
                }
            }
        }
    }

    /**
     * 检查Topic是否存在
     */
    private boolean topicExists(String topicName) {
        try {
            ListTopicsResult result = kafkaAdminClient.listTopics();
            Set<String> topicNames = result.names().get();
            return topicNames.contains(topicName);

        } catch (Exception e) {
            log.error("检查Topic是否存在失败: {}", topicName, e);
            return false;
        }
    }
}

/**
 * Kafka集群监控服务
 */
@Service
public class ClusterMonitorService {

    @Autowired
    private KafkaAdminClient kafkaAdminClient;

    @Autowired
    private JMXMetricsCollector jmxMetricsCollector;

    @Autowired
    private ClusterHealthAnalyzer clusterHealthAnalyzer;

    @Autowired
    private AlertService alertService;

    private final ScheduledExecutorService monitorScheduler;

    public ClusterMonitorService() {
        this.monitorScheduler = Executors.newScheduledThreadPool(5);
    }

    /**
     * 启动集群监控
     */
    public void startMonitoring() {
        // 启动定期监控任务
        monitorScheduler.scheduleAtFixedRate(
            this::monitorCluster,
            0,
            60, // 1分钟
            TimeUnit.SECONDS
        );

        log.info("Kafka集群监控启动成功");
    }

    /**
     * 停止集群监控
     */
    public void stopMonitoring() {
        try {
            monitorScheduler.shutdown();
            if (!monitorScheduler.awaitTermination(30, TimeUnit.SECONDS)) {
                monitorScheduler.shutdownNow();
            }

            log.info("Kafka集群监控停止成功");

        } catch (Exception e) {
            log.error("Kafka集群监控停止失败", e);
        }
    }

    /**
     * 监控集群
     */
    private void monitorCluster() {
        try {
            // 1. 收集集群指标
            ClusterMetrics metrics = collectClusterMetrics();

            // 2. 分析集群健康状态
            ClusterHealthStatus healthStatus = clusterHealthAnalyzer.analyzeHealth(metrics);

            // 3. 检查集群状态
            if (!healthStatus.isHealthy()) {
                handleUnhealthyCluster(healthStatus);
            }

            // 4. 记录监控结果
            recordMonitoringResult(metrics, healthStatus);

        } catch (Exception e) {
            log.error("集群监控失败", e);
        }
    }

    /**
     * 收集集群指标
     */
    private ClusterMetrics collectClusterMetrics() {
        ClusterMetrics metrics = new ClusterMetrics();

        try {
            // 1. 收集基础指标
            collectBasicMetrics(metrics);

            // 2. 收集性能指标
            collectPerformanceMetrics(metrics);

            // 3. 收集存储指标
            collectStorageMetrics(metrics);

            // 4. 收集JMX指标
            collectJMXMetrics(metrics);

        } catch (Exception e) {
            log.error("集群指标收集失败", e);
        }

        return metrics;
    }

    /**
     * 收集基础指标
     */
    private void collectBasicMetrics(ClusterMetrics metrics) {
        try {
            // 获取集群信息
            DescribeClusterResult result = kafkaAdminClient.describeCluster();
            Cluster cluster = result.cluster().get();

            metrics.setClusterId(cluster.clusterResource().clusterId());
            metrics.setBrokerCount(cluster.nodes().size());
            metrics.setControllerId(cluster.controller().id());

            // 获取Topic信息
            ListTopicsResult topicsResult = kafkaAdminClient.listTopics();
            Set<String> topicNames = topicsResult.names().get();
            metrics.setTopicCount(topicNames.size());

        } catch (Exception e) {
            log.error("基础指标收集失败", e);
        }
    }

    /**
     * 收集性能指标
     */
    private void collectPerformanceMetrics(ClusterMetrics metrics) {
        try {
            // 收集JMX性能指标
            Map<String, Object> jmxMetrics = jmxMetricsCollector.collectPerformanceMetrics();

            // 消息生产速率
            Double messagesPerSecond = (Double) jmxMetrics.get("kafka.server:type=BrokerTopicMetrics,name=MessagesInPerSec");
            metrics.setMessagesPerSecond(messagesPerSecond != null ? messagesPerSecond : 0.0);

            // 字节生产速率
            Double bytesPerSecond = (Double) jmxMetrics.get("kafka.server:type=BrokerTopicMetrics,name=BytesInPerSec");
            metrics.setBytesPerSecond(bytesPerSecond != null ? bytesPerSecond : 0.0);

            // 消息消费速率
            Double messagesOutPerSecond = (Double) jmxMetrics.get("kafka.server:type=BrokerTopicMetrics,name=MessagesOutPerSec");
            metrics.setMessagesOutPerSecond(messagesOutPerSecond != null ? messagesOutPerSecond : 0.0);

            // 字节消费速率
            Double bytesOutPerSecond = (Double) jmxMetrics.get("kafka.server:type=BrokerTopicMetrics,name=BytesOutPerSec");
            metrics.setBytesOutPerSecond(bytesOutPerSecond != null ? bytesOutPerSecond : 0.0);

        } catch (Exception e) {
            log.error("性能指标收集失败", e);
        }
    }

    /**
     * 收集存储指标
     */
    private void collectStorageMetrics(ClusterMetrics metrics) {
        try {
            // 收集JMX存储指标
            Map<String, Object> jmxMetrics = jmxMetricsCollector.collectStorageMetrics();

            // 日志大小
            Long logSize = (Long) jmxMetrics.get("kafka.log:type=Log,name=Size");
            metrics.setLogSize(logSize != null ? logSize : 0L);

            // 日志数量
            Long logCount = (Long) jmxMetrics.get("kafka.log:type=Log,name=LogCount");
            metrics.setLogCount(logCount != null ? logCount : 0L);

            // 分区数量
            Long partitionCount = (Long) jmxMetrics.get("kafka.server:type=ReplicaManager,name=PartitionCount");
            metrics.setPartitionCount(partitionCount != null ? partitionCount : 0L);

        } catch (Exception e) {
            log.error("存储指标收集失败", e);
        }
    }

    /**
     * 收集JMX指标
     */
    private void collectJMXMetrics(ClusterMetrics metrics) {
        try {
            // 收集JMX指标
            Map<String, Object> jmxMetrics = jmxMetricsCollector.collectAllMetrics();

            // 设置JMX指标
            metrics.setJmxMetrics(jmxMetrics);

        } catch (Exception e) {
            log.error("JMX指标收集失败", e);
        }
    }

    /**
     * 处理不健康集群
     */
    private void handleUnhealthyCluster(ClusterHealthStatus healthStatus) {
        try {
            // 1. 发送集群告警
            sendClusterAlert(healthStatus);

            // 2. 记录集群问题
            recordClusterIssue(healthStatus);

        } catch (Exception e) {
            log.error("不健康集群处理失败", e);
        }
    }

    /**
     * 发送集群告警
     */
    private void sendClusterAlert(ClusterHealthStatus healthStatus) {
        ClusterAlert alert = new ClusterAlert();
        alert.setAlertType(AlertType.CLUSTER_UNHEALTHY);
        alert.setSeverity(healthStatus.getSeverity());
        alert.setMessage("Kafka集群状态异常");
        alert.setHealthStatus(healthStatus);
        alert.setTimestamp(System.currentTimeMillis());

        // 发送告警
        alertService.sendAlert(alert);
    }

    /**
     * 记录监控结果
     */
    private void recordMonitoringResult(ClusterMetrics metrics, ClusterHealthStatus healthStatus) {
        ClusterMonitoringResult result = new ClusterMonitoringResult();
        result.setTimestamp(System.currentTimeMillis());
        result.setMetrics(metrics);
        result.setHealthStatus(healthStatus);

        // 存储监控结果
        // monitoringResultStorage.store(result);
    }
}