Kafka 4.0移除ZooKeeper后Java客户端连接KRaft：KafkaClient与MetadataManager

Kafka 4.0：移除ZooKeeper后Java客户端连接KRaft架构详解

各位同学，大家好。今天我们来深入探讨Kafka 4.0一个重要的架构变更：移除ZooKeeper，引入KRaft共识机制。同时，我们将重点关注在Java客户端视角下，如何连接到基于KRaft的Kafka集群，以及KafkaClient和MetadataManager这两个关键组件在这一过程中的作用。

1. ZooKeeper的局限性与KRaft的诞生

在Kafka早期版本中，ZooKeeper扮演着集群元数据管理、控制器选举等核心角色。但随着Kafka规模的扩大，ZooKeeper逐渐暴露出一些局限性：

• 性能瓶颈： 大规模集群下，频繁的元数据变更导致ZooKeeper压力巨大，成为性能瓶颈。
• 运维复杂性： 需要独立维护一个ZooKeeper集群，增加了运维成本和复杂度。
• 耦合性： Kafka过度依赖ZooKeeper，导致架构不够简洁。

为了解决这些问题，Kafka社区推出了KRaft（Kafka Raft）共识机制。KRaft将元数据管理和控制器选举等功能集成到Kafka Broker内部，不再依赖ZooKeeper。KRaft的优势在于：

• 性能提升： 降低了对外部系统的依赖，提高了元数据管理的效率。
• 简化运维： 无需维护独立的ZooKeeper集群，降低了运维成本。
• 解耦： 减少了对外部系统的依赖，提高了Kafka的独立性。

2. KRaft架构下的Kafka集群组件

在KRaft架构下，Kafka集群主要包含以下组件：

• Kafka Broker： 负责消息的存储、发送和接收。
• Controller： 负责集群元数据管理、分区 Leader 选举等。在 KRaft 模式下，Controller 与 Broker 运行在同一个进程中，通过配置区分角色。
• Metadata Quorum： 由一组 Controller 组成，负责维护集群元数据。采用 Raft 协议保证元数据的一致性和可用性。

3. Java客户端连接KRaft集群：核心流程

Java客户端连接KRaft集群的核心流程如下：

1. 配置引导服务器列表： 客户端需要配置一组引导服务器（Bootstrap Servers），这些服务器可以是集群中的任何 Broker。
2. 连接引导服务器： 客户端随机选择一个引导服务器建立连接。
3. 获取元数据： 客户端向引导服务器发送请求，获取集群的元数据信息，包括 Broker 地址、分区信息等。
4. 更新元数据： 客户端接收到元数据后，会将其存储在本地的 MetadataManager 中。
5. 连接Broker： 客户端根据元数据信息，连接到相应的 Broker 进行消息的生产和消费。
6. 周期性更新元数据： 客户端会定期向集群发送请求，更新本地的元数据信息，以保持与集群状态的同步。

4. KafkaClient：客户端连接的基石

KafkaClient 是 Kafka Java 客户端的核心接口，它定义了客户端与 Kafka 集群交互的基本方法。KafkaClient 的主要职责包括：

• 建立和管理网络连接： 负责与 Kafka Broker 建立和维护 TCP 连接。
• 发送和接收请求： 负责将客户端的请求序列化成 Kafka 协议格式，并通过网络发送到 Broker。同时，接收 Broker 返回的响应，并进行反序列化。
• 处理连接异常： 负责处理连接断开、超时等异常情况，并进行重试或重新建立连接。

KafkaClient 的实现类通常会采用异步非阻塞 IO 模型，以提高客户端的吞吐量和并发能力。

4.1. KafkaClient 接口定义

public interface KafkaClient {

    /**
     * 发送请求到指定的 Broker。
     *
     * @param destination Broker 的地址。
     * @param request 请求对象。
     * @param requestTimeoutMs 请求超时时间。
     * @return 一个 Future 对象，用于获取响应结果。
     */
    CompletableFuture<ClientResponse> send(Node destination, AbstractRequest.Builder<?> request, int requestTimeoutMs);

    /**
     * 关闭客户端。
     */
    void close();
}

4.2. KafkaClient 实现示例 (简化版)

public class DefaultKafkaClient implements KafkaClient {

    private final SocketAddress address;
    private final Selector selector;
    private SocketChannel channel;

    public DefaultKafkaClient(SocketAddress address) throws IOException {
        this.address = address;
        this.selector = Selector.open();
        this.channel = SocketChannel.open();
        this.channel.configureBlocking(false);
        this.channel.connect(address);
        this.channel.register(selector, SelectionKey.OP_CONNECT);
    }

    @Override
    public CompletableFuture<ClientResponse> send(Node destination, AbstractRequest.Builder<?> request, int requestTimeoutMs) {
        // 省略：序列化请求、发送数据、接收响应等逻辑
        return CompletableFuture.completedFuture(new ClientResponse(new RequestHeader(request.apiKey(), request.version(), "client", 123),new Struct(new Schema(new Field("test",Type.STRING))),Clock.SYSTEM.currentTimeMillis(),null));
    }

    @Override
    public void close() {
        try {
            channel.close();
            selector.close();
        } catch (IOException e) {
            // 处理异常
        }
    }
}

注意： 上面的代码只是一个简化版的 KafkaClient 实现，实际的实现会更加复杂，包括连接池管理、请求重试、错误处理等。

5. MetadataManager：元数据管理的中心

MetadataManager 是 Kafka Java 客户端中负责管理集群元数据的核心组件。它维护了集群的 Broker 信息、分区信息、Topic 信息等。MetadataManager 的主要职责包括：

• 存储元数据： 将从 Kafka 集群获取的元数据信息存储在本地缓存中。
• 更新元数据： 定期向 Kafka 集群发送请求，更新本地的元数据信息。
• 提供元数据查询接口： 提供接口供客户端查询 Broker 地址、分区 Leader 等信息。
• 管理监听器： 允许客户端注册监听器，以便在元数据发生变化时得到通知。

MetadataManager 的设计目标是提供一个高效、可靠的元数据管理机制，以便客户端能够快速地定位到目标 Broker，并进行消息的生产和消费。

5.1. MetadataManager 接口定义

public interface MetadataManager {

    /**
     * 获取指定 Topic 的分区信息。
     *
     * @param topic Topic 名称。
     * @return 分区信息列表。
     */
    List<PartitionInfo> getPartitions(String topic);

    /**
     * 获取指定分区的 Leader Broker。
     *
     * @param topic Topic 名称。
     * @param partition 分区 ID。
     * @return Leader Broker 的地址。
     */
    Node getPartitionLeader(String topic, int partition);

    /**
     * 更新元数据。
     *
     * @param metadata 元数据信息。
     */
    void update(MetadataResponse metadata);

    /**
     * 注册元数据监听器。
     *
     * @param listener 元数据监听器。
     */
    void addListener(Listener listener);

    /**
     * 移除元数据监听器。
     *
     * @param listener 元数据监听器.
     */
    void removeListener(Listener listener);

    interface Listener {
        void onMetadataUpdate(Metadata metadata);
    }

    /**
     * 关闭 MetadataManager
     */
    void close();
}

5.2. MetadataManager 实现示例 (简化版)

public class DefaultMetadataManager implements MetadataManager {

    private final Map<String, List<PartitionInfo>> topicPartitions = new ConcurrentHashMap<>();
    private final List<Listener> listeners = new CopyOnWriteArrayList<>();
    private final KafkaClient kafkaClient;
    private final List<String> bootstrapServers;
    private final long refreshIntervalMs;
    private volatile boolean running = true;

    public DefaultMetadataManager(List<String> bootstrapServers, KafkaClient kafkaClient, long refreshIntervalMs) {
        this.bootstrapServers = bootstrapServers;
        this.kafkaClient = kafkaClient;
        this.refreshIntervalMs = refreshIntervalMs;
        startMetadataUpdater();
    }

    @Override
    public List<PartitionInfo> getPartitions(String topic) {
        return topicPartitions.get(topic);
    }

    @Override
    public Node getPartitionLeader(String topic, int partition) {
        List<PartitionInfo> partitions = topicPartitions.get(topic);
        if (partitions != null) {
            for (PartitionInfo partitionInfo : partitions) {
                if (partitionInfo.partition() == partition) {
                    return partitionInfo.leader();
                }
            }
        }
        return null;
    }

    @Override
    public void update(MetadataResponse metadata) {
        Map<String, List<PartitionInfo>> newTopicPartitions = new HashMap<>();
        for (TopicMetadata topicMetadata : metadata.topicMetadata()) {
            String topic = topicMetadata.topic();
            List<PartitionInfo> partitionInfos = new ArrayList<>();
            for (PartitionMetadata partitionMetadata : topicMetadata.partitions()) {
                partitionInfos.add(new PartitionInfo(topic, partitionMetadata.partition(), partitionMetadata.leader(), partitionMetadata.replicas(), partitionMetadata.isr()));
            }
            newTopicPartitions.put(topic, partitionInfos);
        }
        topicPartitions.clear();
        topicPartitions.putAll(newTopicPartitions);

        listeners.forEach(listener -> listener.onMetadataUpdate(new Metadata(metadata.clusterId(), metadata.controller(), metadata.topicMetadata(), Clock.SYSTEM.currentTimeMillis())));
    }

    @Override
    public void addListener(Listener listener) {
        this.listeners.add(listener);
    }

    @Override
    public void removeListener(Listener listener) {
        this.listeners.remove(listener);
    }

    @Override
    public void close() {
        running = false;
    }

    private void startMetadataUpdater() {
        new Thread(() -> {
            while (running) {
                try {
                    updateMetadata();
                    Thread.sleep(refreshIntervalMs);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                    break;
                } catch (Exception e) {
                    // 处理异常
                }
            }
        }).start();
    }

    private void updateMetadata() {
        bootstrapServers.forEach(bootstrapServer -> {
            String[] hostPort = bootstrapServer.split(":");
            String host = hostPort[0];
            int port = Integer.parseInt(hostPort[1]);
            Node node = new Node(1, host, port);

            MetadataRequest.Builder builder = new MetadataRequest.Builder(new ArrayList<>());
            try {
                CompletableFuture<ClientResponse> future = kafkaClient.send(node, builder, 10000);
                future.whenComplete((response, throwable) -> {
                    if (throwable != null) {
                        // 处理异常
                    } else {
                        MetadataResponse metadataResponse = (MetadataResponse) response.responseBody();
                        update(metadataResponse);
                    }
                });
            } catch (Exception e) {
                // 处理异常
            }
        });
    }
}

注意： 上面的代码只是一个简化版的 MetadataManager 实现，实际的实现会更加复杂，包括缓存失效策略、错误处理、并发控制等。

6. 从ZooKeeper到KRaft：元数据获取的差异

在ZooKeeper架构下，客户端通过ZooKeeper获取集群元数据。而在KRaft架构下，客户端直接向Broker发送请求获取元数据。这意味着，客户端需要知道至少一个Broker的地址，以便进行引导。

特性	ZooKeeper	KRaft
元数据存储	ZooKeeper 集群	Kafka Broker (Controller)
元数据获取方式	通过 ZooKeeper 客户端 API 获取	通过 Kafka 客户端向 Broker 发送请求获取
引导服务器	ZooKeeper 集群地址列表	Kafka Broker 地址列表
依赖性	强依赖 ZooKeeper	不依赖 ZooKeeper

7. 完整示例：连接KRaft集群并生产消息

下面是一个完整的示例，演示了如何使用 Java 客户端连接到 KRaft 集群，并生产一条消息：

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

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

public class KRaftProducerExample {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 1. 配置 KafkaProducer
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092"); // KRaft 集群的 Broker 地址
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 2. 创建 KafkaProducer
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        // 3. 创建 ProducerRecord
        String topic = "my-topic";
        String key = "key1";
        String value = "value1";
        ProducerRecord<String, String> record = new ProducerRecord<>(topic, key, value);

        // 4. 发送消息
        try {
            producer.send(record).get(); // 同步发送
            System.out.println("消息发送成功！");
        } catch (Exception e) {
            System.err.println("消息发送失败：" + e.getMessage());
        } finally {
            // 5. 关闭 KafkaProducer
            producer.close();
        }
    }
}

注意：

• 需要将 BOOTSTRAP_SERVERS_CONFIG 配置为 KRaft 集群的 Broker 地址。
• 示例中使用的是同步发送方式，也可以使用异步发送方式。

8. 客户端配置的关键参数

连接KRaft集群，客户端需要配置一些关键参数：

参数名称	描述
bootstrap.servers	指定 Kafka 集群的 Broker 地址列表。客户端会从中选择一个 Broker 建立连接，并获取集群的元数据信息。
metadata.max.age.ms	指定元数据的最大过期时间。客户端会定期向 Kafka 集群发送请求，更新本地的元数据信息。如果元数据超过指定时间未更新，客户端会强制刷新元数据。
client.id	指定客户端的 ID。用于在 Kafka Broker 上区分不同的客户端。
request.timeout.ms	指定客户端发送请求到 Kafka Broker 的超时时间。
retry.backoff.ms	指定客户端在重试请求时，等待的时间间隔。
connections.max.idle.ms	指定客户端与 Kafka Broker 之间的连接，在空闲状态下的最大保持时间。超过指定时间，连接会被关闭。
security.protocol	指定客户端与 Kafka Broker 之间的安全协议。常用的安全协议包括 PLAINTEXT、SSL、SASL_PLAINTEXT、SASL_SSL。
sasl.mechanism	当 security.protocol 配置为 SASL_PLAINTEXT 或 SASL_SSL 时，需要指定 SASL 机制。常用的 SASL 机制包括 GSSAPI、PLAIN、SCRAM-SHA-256、SCRAM-SHA-512。
sasl.jaas.config	当 security.protocol 配置为 SASL_PLAINTEXT 或 SASL_SSL 时，并且 SASL 机制需要 JAAS 配置时，需要指定 JAAS 配置文件的路径。

9. 总结

Kafka 4.0 移除 ZooKeeper 并引入 KRaft 共识机制，简化了架构，提高了性能和可维护性。Java 客户端连接 KRaft 集群的关键在于正确配置引导服务器列表，并通过 KafkaClient 和 MetadataManager 这两个组件进行元数据管理和网络通信。理解这两个组件的工作原理，能够帮助我们更好地使用 Kafka Java 客户端，构建高效、可靠的 Kafka 应用。

10. KRaft架构下的开发新视角

从开发者的角度来看，Kafka 4.0 的 KRaft 架构简化了配置和部署流程。不再需要额外的 ZooKeeper 集群，降低了运维成本。然而，开发者需要更加关注 Broker 的配置和管理，确保 Controller 能够正常运行，维护集群的元数据。同时，也需要关注客户端的配置，特别是引导服务器列表，确保客户端能够正确地连接到 Kafka 集群。