第97集Kafka消息队列运维监控与优化实战

/**
 * Kafka集群监控数据实体类
 */
@Data
@TableName("kafka_cluster_monitor")
public class KafkaClusterMonitor {
    
    @TableId(type = IdType.AUTO)
    private Long id;                    // 主键ID
    
    private String clusterName;          // 集群名称
    
    private String hostname;             // 主机名
    
    private String ip;                   // IP地址
    
    private Integer brokerCount;         // Broker数量
    
    private Integer topicCount;          // Topic数量
    
    private Integer partitionCount;      // 分区数量
    
    private Integer replicaCount;        // 副本数量
    
    private Long totalMessages;          // 总消息数
    
    private Long totalBytes;             // 总字节数
    
    private Double throughput;           // 吞吐量(消息/秒)
    
    private Double avgLatency;           // 平均延迟(毫秒)
    
    private Integer underReplicatedPartitions; // 未充分复制分区数
    
    private Integer offlinePartitions;   // 离线分区数
    
    private String clusterStatus;        // 集群状态
    
    private Date collectTime;            // 采集时间
    
    private Date createTime;             // 创建时间
}

/**
 * Kafka Topic监控数据实体类
 */
@Data
@TableName("kafka_topic_monitor")
public class KafkaTopicMonitor {
    
    @TableId(type = IdType.AUTO)
    private Long id;                    // 主键ID
    
    private String clusterName;          // 集群名称
    
    private String topicName;            // Topic名称
    
    private Integer partitionCount;      // 分区数量
    
    private Integer replicaCount;        // 副本数量
    
    private Long messageCount;          // 消息数量
    
    private Long logSize;               // 日志大小
    
    private Double messageRate;         // 消息速率(消息/秒)
    
    private Double byteRate;             // 字节速率(字节/秒)
    
    private Long consumerLag;           // 消费者延迟
    
    private Integer underReplicatedPartitions; // 未充分复制分区数
    
    private Integer offlinePartitions;   // 离线分区数
    
    private String topicStatus;         // Topic状态
    
    private Date collectTime;            // 采集时间
    
    private Date createTime;             // 创建时间
}

/**
 * Kafka监控服务
 * 负责Kafka集群和Topic数据的采集、存储和分析
 */
@Service
public class KafkaMonitorService {

    @Autowired
    private KafkaClusterMonitorMapper kafkaClusterMonitorMapper;
    
    @Autowired
    private KafkaTopicMonitorMapper kafkaTopicMonitorMapper;
    
    @Autowired
    private RedisTemplate<String, Object> redisTemplate;
    
    @Autowired
    private AlertService alertService;
    
    private AdminClient adminClient;
    private KafkaConsumer<String, String> consumer;

    /**
     * 初始化Kafka客户端
     */
    @PostConstruct
    public void initKafkaClients() {
        try {
            // 创建AdminClient
            Properties adminProps = new Properties();
            adminProps.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
            adminProps.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 30000);
            adminClient = AdminClient.create(adminProps);
            
            // 创建Consumer
            Properties consumerProps = new Properties();
            consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
            consumerProps.put(ConsumerConfig.GROUP_ID_CONFIG, "kafka-monitor-group");
            consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
            consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
            consumerProps.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
            consumer = new KafkaConsumer<>(consumerProps);
            
            log.info("Kafka客户端初始化成功");
            
        } catch (Exception e) {
            log.error("初始化Kafka客户端失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 采集Kafka集群数据
     * 定期采集Kafka集群和Topic信息
     */
    @Scheduled(fixedRate = 10000) // 每10秒执行一次
    public void collectKafkaData() {
        try {
            // 1. 采集集群信息
            collectClusterInfo();
            
            // 2. 采集Topic信息
            collectTopicInfo();
            
        } catch (Exception e) {
            log.error("采集Kafka数据失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 采集集群信息
     */
    private void collectClusterInfo() {
        try {
            // 1. 获取集群信息
            KafkaClusterInfo clusterInfo = getClusterInfo();
            
            // 2. 创建集群监控数据
            KafkaClusterMonitor monitorData = createClusterMonitorData(clusterInfo);
            
            // 3. 保存到数据库
            kafkaClusterMonitorMapper.insert(monitorData);
            
            // 4. 更新缓存
            updateClusterCache(monitorData);
            
            // 5. 检查集群告警
            checkClusterAlert(monitorData);
            
            log.debug("采集集群信息: brokerCount={}, topicCount={}", 
                    monitorData.getBrokerCount(), monitorData.getTopicCount());
                    
        } catch (Exception e) {
            log.error("采集集群信息失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 获取集群信息
     */
    private KafkaClusterInfo getClusterInfo() {
        KafkaClusterInfo clusterInfo = new KafkaClusterInfo();
        
        try {
            // 获取Broker信息
            DescribeClusterResult clusterResult = adminClient.describeCluster();
            Collection<Node> nodes = clusterResult.nodes().get();
            clusterInfo.setBrokerCount(nodes.size());
            
            // 获取Topic信息
            ListTopicsResult topicsResult = adminClient.listTopics();
            Set<String> topicNames = topicsResult.names().get();
            clusterInfo.setTopicCount(topicNames.size());
            
            // 获取分区和副本信息
            int totalPartitions = 0;
            int totalReplicas = 0;
            int underReplicatedPartitions = 0;
            int offlinePartitions = 0;
            
            for (String topicName : topicNames) {
                DescribeTopicsResult topicsDescResult = adminClient.describeTopics(Arrays.asList(topicName));
                Map<String, TopicDescription> topicDescriptions = topicsDescResult.all().get();
                
                TopicDescription topicDescription = topicDescriptions.get(topicName);
                if (topicDescription != null) {
                    totalPartitions += topicDescription.partitions().size();
                    
                    for (TopicPartitionInfo partitionInfo : topicDescription.partitions()) {
                        totalReplicas += partitionInfo.replicas().size();
                        
                        // 检查未充分复制的分区
                        if (partitionInfo.replicas().size() < partitionInfo.replicas().size()) {
                            underReplicatedPartitions++;
                        }
                        
                        // 检查离线分区
                        if (partitionInfo.leader() == null) {
                            offlinePartitions++;
                        }
                    }
                }
            }
            
            clusterInfo.setPartitionCount(totalPartitions);
            clusterInfo.setReplicaCount(totalReplicas);
            clusterInfo.setUnderReplicatedPartitions(underReplicatedPartitions);
            clusterInfo.setOfflinePartitions(offlinePartitions);
            
            // 计算吞吐量和延迟
            clusterInfo.setThroughput(calculateThroughput());
            clusterInfo.setAvgLatency(calculateAvgLatency());
            
            // 设置集群状态
            clusterInfo.setClusterStatus(determineClusterStatus(clusterInfo));
            
        } catch (Exception e) {
            log.error("获取集群信息失败: {}", e.getMessage(), e);
        }
        
        return clusterInfo;
    }

    /**
     * 计算吞吐量
     */
    private double calculateThroughput() {
        try {
            // 简化实现，实际应该从JMX或Kafka监控工具获取
            return 1000.0; // 1000 消息/秒
        } catch (Exception e) {
            return 0.0;
        }
    }

    /**
     * 计算平均延迟
     */
    private double calculateAvgLatency() {
        try {
            // 简化实现，实际应该从JMX或Kafka监控工具获取
            return 10.0; // 10ms
        } catch (Exception e) {
            return 0.0;
        }
    }

    /**
     * 确定集群状态
     */
    private String determineClusterStatus(KafkaClusterInfo clusterInfo) {
        String status = "Healthy";
        
        if (clusterInfo.getUnderReplicatedPartitions() > 0) {
            status = "UnderReplicated";
        } else if (clusterInfo.getOfflinePartitions() > 0) {
            status = "OfflinePartitions";
        } else if (clusterInfo.getThroughput() < 100) {
            status = "LowThroughput";
        }
        
        return status;
    }

    /**
     * 创建集群监控数据
     */
    private KafkaClusterMonitor createClusterMonitorData(KafkaClusterInfo clusterInfo) {
        KafkaClusterMonitor monitorData = new KafkaClusterMonitor();
        
        // 设置基本信息
        monitorData.setClusterName(clusterInfo.getClusterName());
        monitorData.setHostname(getHostname());
        monitorData.setIp(getLocalIpAddress());
        monitorData.setCollectTime(new Date());
        monitorData.setCreateTime(new Date());
        
        // 设置集群信息
        monitorData.setBrokerCount(clusterInfo.getBrokerCount());
        monitorData.setTopicCount(clusterInfo.getTopicCount());
        monitorData.setPartitionCount(clusterInfo.getPartitionCount());
        monitorData.setReplicaCount(clusterInfo.getReplicaCount());
        
        // 设置性能信息
        monitorData.setThroughput(clusterInfo.getThroughput());
        monitorData.setAvgLatency(clusterInfo.getAvgLatency());
        
        // 设置状态信息
        monitorData.setUnderReplicatedPartitions(clusterInfo.getUnderReplicatedPartitions());
        monitorData.setOfflinePartitions(clusterInfo.getOfflinePartitions());
        monitorData.setClusterStatus(clusterInfo.getClusterStatus());
        
        return monitorData;
    }

    /**
     * 采集Topic信息
     */
    private void collectTopicInfo() {
        try {
            // 获取所有Topic
            ListTopicsResult topicsResult = adminClient.listTopics();
            Set<String> topicNames = topicsResult.names().get();
            
            for (String topicName : topicNames) {
                try {
                    // 获取Topic详细信息
                    DescribeTopicsResult topicsDescResult = adminClient.describeTopics(Arrays.asList(topicName));
                    Map<String, TopicDescription> topicDescriptions = topicsDescResult.all().get();
                    
                    TopicDescription topicDescription = topicDescriptions.get(topicName);
                    if (topicDescription != null) {
                        // 创建Topic监控数据
                        KafkaTopicMonitor topicMonitor = createTopicMonitorData(topicDescription);
                        
                        // 保存到数据库
                        kafkaTopicMonitorMapper.insert(topicMonitor);
                        
                        // 更新缓存
                        updateTopicCache(topicMonitor);
                        
                        // 检查Topic告警
                        checkTopicAlert(topicMonitor);
                    }
                    
                } catch (Exception e) {
                    log.error("处理Topic失败: topicName={}, error={}", 
                            topicName, e.getMessage());
                }
            }
            
        } catch (Exception e) {
            log.error("采集Topic信息失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 创建Topic监控数据
     */
    private KafkaTopicMonitor createTopicMonitorData(TopicDescription topicDescription) {
        KafkaTopicMonitor topicMonitor = new KafkaTopicMonitor();
        
        // 设置基本信息
        topicMonitor.setClusterName("production-cluster");
        topicMonitor.setTopicName(topicDescription.name());
        topicMonitor.setCollectTime(new Date());
        topicMonitor.setCreateTime(new Date());
        
        // 设置分区和副本信息
        topicMonitor.setPartitionCount(topicDescription.partitions().size());
        
        int replicaCount = 0;
        int underReplicatedPartitions = 0;
        int offlinePartitions = 0;
        
        for (TopicPartitionInfo partitionInfo : topicDescription.partitions()) {
            replicaCount += partitionInfo.replicas().size();
            
            // 检查未充分复制的分区
            if (partitionInfo.replicas().size() < partitionInfo.replicas().size()) {
                underReplicatedPartitions++;
            }
            
            // 检查离线分区
            if (partitionInfo.leader() == null) {
                offlinePartitions++;
            }
        }
        
        topicMonitor.setReplicaCount(replicaCount);
        topicMonitor.setUnderReplicatedPartitions(underReplicatedPartitions);
        topicMonitor.setOfflinePartitions(offlinePartitions);
        
        // 设置性能信息
        topicMonitor.setMessageRate(calculateTopicMessageRate(topicDescription.name()));
        topicMonitor.setByteRate(calculateTopicByteRate(topicDescription.name()));
        topicMonitor.setConsumerLag(calculateConsumerLag(topicDescription.name()));
        
        // 设置状态信息
        topicMonitor.setTopicStatus(determineTopicStatus(topicMonitor));
        
        return topicMonitor;
    }

    /**
     * 计算Topic消息速率
     */
    private double calculateTopicMessageRate(String topicName) {
        try {
            // 简化实现，实际应该从JMX或Kafka监控工具获取
            return 100.0; // 100 消息/秒
        } catch (Exception e) {
            return 0.0;
        }
    }

    /**
     * 计算Topic字节速率
     */
    private double calculateTopicByteRate(String topicName) {
        try {
            // 简化实现，实际应该从JMX或Kafka监控工具获取
            return 1024.0; // 1024 字节/秒
        } catch (Exception e) {
            return 0.0;
        }
    }

    /**
     * 计算消费者延迟
     */
    private long calculateConsumerLag(String topicName) {
        try {
            // 简化实现，实际应该从消费者组信息获取
            return 0;
        } catch (Exception e) {
            return 0;
        }
    }

    /**
     * 确定Topic状态
     */
    private String determineTopicStatus(KafkaTopicMonitor topicMonitor) {
        String status = "Healthy";
        
        if (topicMonitor.getUnderReplicatedPartitions() > 0) {
            status = "UnderReplicated";
        } else if (topicMonitor.getOfflinePartitions() > 0) {
            status = "OfflinePartitions";
        } else if (topicMonitor.getConsumerLag() > 10000) {
            status = "HighConsumerLag";
        }
        
        return status;
    }

    /**
     * 更新集群缓存
     */
    private void updateClusterCache(KafkaClusterMonitor monitorData) {
        try {
            String cacheKey = "kafka:cluster:" + monitorData.getClusterName();
            redisTemplate.opsForValue().set(cacheKey, monitorData, Duration.ofMinutes(5));
            
        } catch (Exception e) {
            log.warn("更新集群缓存失败: {}", e.getMessage());
        }
    }

    /**
     * 更新Topic缓存
     */
    private void updateTopicCache(KafkaTopicMonitor topicMonitor) {
        try {
            String cacheKey = "kafka:topic:" + topicMonitor.getTopicName();
            redisTemplate.opsForValue().set(cacheKey, topicMonitor, Duration.ofMinutes(5));
            
        } catch (Exception e) {
            log.warn("更新Topic缓存失败: {}", e.getMessage());
        }
    }

    /**
     * 检查集群告警
     */
    private void checkClusterAlert(KafkaClusterMonitor monitorData) {
        try {
            String alertType = null;
            String alertLevel = null;
            String alertMessage = null;
            
            // 检查未充分复制分区告警
            if (monitorData.getUnderReplicatedPartitions() > 0) {
                alertType = "KAFKA_UNDER_REPLICATED_PARTITIONS";
                alertLevel = "WARNING";
                alertMessage = String.format("Kafka未充分复制分区: %d", monitorData.getUnderReplicatedPartitions());
            }
            
            // 检查离线分区告警
            if (monitorData.getOfflinePartitions() > 0) {
                alertType = "KAFKA_OFFLINE_PARTITIONS";
                alertLevel = "CRITICAL";
                alertMessage = String.format("Kafka离线分区: %d", monitorData.getOfflinePartitions());
            }
            
            // 检查吞吐量告警
            if (monitorData.getThroughput() < 100) {
                alertType = "KAFKA_LOW_THROUGHPUT";
                alertLevel = "WARNING";
                alertMessage = String.format("Kafka吞吐量过低: %.2f", monitorData.getThroughput());
            }
            
            // 发送告警
            if (alertType != null) {
                sendClusterAlert(monitorData, alertType, alertLevel, alertMessage);
            }
            
        } catch (Exception e) {
            log.error("检查集群告警失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 检查Topic告警
     */
    private void checkTopicAlert(KafkaTopicMonitor topicMonitor) {
        try {
            String alertType = null;
            String alertLevel = null;
            String alertMessage = null;
            
            // 检查未充分复制分区告警
            if (topicMonitor.getUnderReplicatedPartitions() > 0) {
                alertType = "KAFKA_TOPIC_UNDER_REPLICATED";
                alertLevel = "WARNING";
                alertMessage = String.format("Topic未充分复制分区: %s, 数量: %d", 
                        topicMonitor.getTopicName(), topicMonitor.getUnderReplicatedPartitions());
            }
            
            // 检查离线分区告警
            if (topicMonitor.getOfflinePartitions() > 0) {
                alertType = "KAFKA_TOPIC_OFFLINE";
                alertLevel = "CRITICAL";
                alertMessage = String.format("Topic离线分区: %s, 数量: %d", 
                        topicMonitor.getTopicName(), topicMonitor.getOfflinePartitions());
            }
            
            // 检查消费者延迟告警
            if (topicMonitor.getConsumerLag() > 10000) {
                alertType = "KAFKA_TOPIC_CONSUMER_LAG";
                alertLevel = "WARNING";
                alertMessage = String.format("Topic消费者延迟过高: %s, 延迟: %d", 
                        topicMonitor.getTopicName(), topicMonitor.getConsumerLag());
            }
            
            // 发送告警
            if (alertType != null) {
                sendTopicAlert(topicMonitor, alertType, alertLevel, alertMessage);
            }
            
        } catch (Exception e) {
            log.error("检查Topic告警失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 发送集群告警
     */
    private void sendClusterAlert(KafkaClusterMonitor monitorData, String alertType, String alertLevel, String alertMessage) {
        try {
            String alertKey = "kafka:cluster:alert:" + monitorData.getClusterName() + ":" + alertType;
            Boolean hasAlert = redisTemplate.hasKey(alertKey);
            
            if (hasAlert == null || !hasAlert) {
                AlertMessage alert = new AlertMessage();
                alert.setType(alertType);
                alert.setLevel(alertLevel);
                alert.setMessage(alertMessage);
                alert.setTimestamp(new Date());
                alert.setHostname(monitorData.getHostname());
                
                alertService.sendAlert(alert);
                
                redisTemplate.opsForValue().set(alertKey, "1", Duration.ofMinutes(5));
                
                log.warn("发送集群告警: clusterName={}, type={}, level={}", 
                        monitorData.getClusterName(), alertType, alertLevel);
            }
            
        } catch (Exception e) {
            log.error("发送集群告警失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 发送Topic告警
     */
    private void sendTopicAlert(KafkaTopicMonitor topicMonitor, String alertType, String alertLevel, String alertMessage) {
        try {
            String alertKey = "kafka:topic:alert:" + topicMonitor.getTopicName() + ":" + alertType;
            Boolean hasAlert = redisTemplate.hasKey(alertKey);
            
            if (hasAlert == null || !hasAlert) {
                AlertMessage alert = new AlertMessage();
                alert.setType(alertType);
                alert.setLevel(alertLevel);
                alert.setMessage(alertMessage);
                alert.setTimestamp(new Date());
                alert.setHostname(topicMonitor.getClusterName());
                
                alertService.sendAlert(alert);
                
                redisTemplate.opsForValue().set(alertKey, "1", Duration.ofMinutes(5));
                
                log.warn("发送Topic告警: topicName={}, type={}, level={}", 
                        topicMonitor.getTopicName(), alertType, alertLevel);
            }
            
        } catch (Exception e) {
            log.error("发送Topic告警失败: {}", e.getMessage(), e);
        }
    }

    /**
     * 获取实时集群数据
     */
    public KafkaClusterMonitor getRealTimeClusterData(String clusterName) {
        String cacheKey = "kafka:cluster:" + clusterName;
        return (KafkaClusterMonitor) redisTemplate.opsForValue().get(cacheKey);
    }

    /**
     * 获取实时Topic数据
     */
    public KafkaTopicMonitor getRealTimeTopicData(String topicName) {
        String cacheKey = "kafka:topic:" + topicName;
        return (KafkaTopicMonitor) redisTemplate.opsForValue().get(cacheKey);
    }

    /**
     * 获取主机名
     */
    private String getHostname() {
        try {
            return InetAddress.getLocalHost().getHostName();
        } catch (UnknownHostException e) {
            return "unknown";
        }
    }

    /**
     * 获取本地IP地址
     */
    private String getLocalIpAddress() {
        try {
            return InetAddress.getLocalHost().getHostAddress();
        } catch (UnknownHostException e) {
            return "127.0.0.1";
        }
    }

    /**
     * 关闭Kafka客户端
     */
    @PreDestroy
    public void shutdownKafkaClients() {
        try {
            if (adminClient != null) {
                adminClient.close();
            }
            
            if (consumer != null) {
                consumer.close();
            }
            
            log.info("Kafka客户端关闭成功");
            
        } catch (Exception e) {
            log.error("关闭Kafka客户端失败: {}", e.getMessage(), e);
        }
    }
}

/**
 * Kafka集群信息实体类
 */
@Data
public class KafkaClusterInfo {
    private String clusterName;          // 集群名称
    private Integer brokerCount;         // Broker数量
    private Integer topicCount;          // Topic数量
    private Integer partitionCount;      // 分区数量
    private Integer replicaCount;        // 副本数量
    private Integer underReplicatedPartitions; // 未充分复制分区数
    private Integer offlinePartitions;   // 离线分区数
    private Double throughput;           // 吞吐量
    private Double avgLatency;           // 平均延迟
    private String clusterStatus;        // 集群状态
}

/**
 * Kafka管理服务
 * 提供Kafka集群管理功能
 */
@Service
public class KafkaManagementService {

    @Autowired
    private AdminClient adminClient;
    
    @Autowired
    private AlertService alertService;

    /**
     * 创建Topic
     */
    public void createTopic(String topicName, int partitions, int replicationFactor) {
        try {
            // 创建Topic配置
            NewTopic newTopic = new NewTopic(topicName, partitions, (short) replicationFactor);
            
            // 创建Topic
            CreateTopicsResult result = adminClient.createTopics(Arrays.asList(newTopic));
            result.all().get();
            
            log.info("创建Topic成功: topicName={}, partitions={}, replicationFactor={}", 
                    topicName, partitions, replicationFactor);
            
        } catch (Exception e) {
            log.error("创建Topic失败: topicName={}, error={}", topicName, e.getMessage(), e);
        }
    }

    /**
     * 删除Topic
     */
    public void deleteTopic(String topicName) {
        try {
            // 删除Topic
            DeleteTopicsResult result = adminClient.deleteTopics(Arrays.asList(topicName));
            result.all().get();
            
            log.info("删除Topic成功: topicName={}", topicName);
            
        } catch (Exception e) {
            log.error("删除Topic失败: topicName={}, error={}", topicName, e.getMessage(), e);
        }
    }

    /**
     * 增加Topic分区
     */
    public void increaseTopicPartitions(String topicName, int newPartitionCount) {
        try {
            // 创建分区配置
            Map<String, NewPartitions> newPartitionsMap = new HashMap<>();
            newPartitionsMap.put(topicName, NewPartitions.increaseTo(newPartitionCount));
            
            // 增加分区
            CreatePartitionsResult result = adminClient.createPartitions(newPartitionsMap);
            result.all().get();
            
            log.info("增加Topic分区成功: topicName={}, newPartitionCount={}", 
                    topicName, newPartitionCount);
            
        } catch (Exception e) {
            log.error("增加Topic分区失败: topicName={}, error={}", topicName, e.getMessage(), e);
        }
    }

    /**
     * 获取Topic详细信息
     */
    public TopicDescription getTopicDescription(String topicName) {
        try {
            DescribeTopicsResult result = adminClient.describeTopics(Arrays.asList(topicName));
            Map<String, TopicDescription> topicDescriptions = result.all().get();
            
            return topicDescriptions.get(topicName);
            
        } catch (Exception e) {
            log.error("获取Topic详细信息失败: topicName={}, error={}", topicName, e.getMessage(), e);
            return null;
        }
    }

    /**
     * 获取所有Topic
     */
    public Set<String> getAllTopics() {
        try {
            ListTopicsResult result = adminClient.listTopics();
            return result.names().get();
            
        } catch (Exception e) {
            log.error("获取所有Topic失败: {}", e.getMessage(), e);
            return new HashSet<>();
        }
    }

    /**
     * 获取消费者组信息
     */
    public List<ConsumerGroupDescription> getConsumerGroups() {
        try {
            ListConsumerGroupsResult result = adminClient.listConsumerGroups();
            Collection<ConsumerGroupListing> consumerGroupListings = result.all().get();
            
            List<String> groupIds = consumerGroupListings.stream()
                    .map(ConsumerGroupListing::groupId)
                    .collect(Collectors.toList());
            
            DescribeConsumerGroupsResult describeResult = adminClient.describeConsumerGroups(groupIds);
            Map<String, ConsumerGroupDescription> consumerGroupDescriptions = describeResult.all().get();
            
            return new ArrayList<>(consumerGroupDescriptions.values());
            
        } catch (Exception e) {
            log.error("获取消费者组信息失败: {}", e.getMessage(), e);
            return new ArrayList<>();
        }
    }

    /**
     * 重置消费者组偏移量
     */
    public void resetConsumerGroupOffset(String groupId, String topicName, long offset) {
        try {
            // 创建偏移量配置
            Map<TopicPartition, OffsetAndMetadata> offsetMap = new HashMap<>();
            TopicPartition topicPartition = new TopicPartition(topicName, 0);
            offsetMap.put(topicPartition, new OffsetAndMetadata(offset));
            
            // 重置偏移量
            AlterConsumerGroupOffsetsResult result = adminClient.alterConsumerGroupOffsets(groupId, offsetMap);
            result.all().get();
            
            log.info("重置消费者组偏移量成功: groupId={}, topicName={}, offset={}", 
                    groupId, topicName, offset);
            
        } catch (Exception e) {
            log.error("重置消费者组偏移量失败: groupId={}, topicName={}, error={}", 
                    groupId, topicName, e.getMessage(), e);
        }
    }

    /**
     * 获取Broker信息
     */
    public Collection<Node> getBrokers() {
        try {
            DescribeClusterResult result = adminClient.describeCluster();
            return result.nodes().get();
            
        } catch (Exception e) {
            log.error("获取Broker信息失败: {}", e.getMessage(), e);
            return new ArrayList<>();
        }
    }

    /**
     * 获取集群配置
     */
    public Config getClusterConfig() {
        try {
            DescribeConfigsResult result = adminClient.describeConfigs(
                    Arrays.asList(new ConfigResource(ConfigResource.Type.BROKER, "0")));
            Map<ConfigResource, Config> configs = result.all().get();
            
            return configs.values().iterator().next();
            
        } catch (Exception e) {
            log.error("获取集群配置失败: {}", e.getMessage(), e);
            return null;
        }
    }
}