当日志结束时，我如何停止尝试消费Kafka消息？

// 创建消费者实例
KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);

// 订阅主题
consumer.subscribe(Collections.singletonList("my-topic"));

// 设置日志结束的条件
boolean isRunning = true;

try {
    while (isRunning) {
        // 拉取消息并进行处理
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, String> record : records) {
            // 处理消息的逻辑
            // ...
        }
        
        // 检查日志是否结束
        if (isLogEnd()) {
            // 停止轮询
            isRunning = false;
            
            // 继续处理剩余消息
            while (!consumer.isEmpty()) {
                ConsumerRecords<String, String> remainingRecords = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : remainingRecords) {
                    // 处理剩余消息的逻辑
                    // ...
                }
            }
        }
    }
} finally {
    // 关闭消费者
    consumer.close();
}