clients

Active contributors: Andrew Schofield, Lianet Magrans, Ken Huang, PoAn Yang, Kirk True

Purpose

The clients/ module contains the Java producer, consumer, admin, and share-consumer clients, plus the shared common/* types (configs, errors, the wire protocol, metrics, security, serialization) that everything else depends on. Anything that talks to a Kafka broker — a user application, a Streams app, a Connect worker, the broker itself for inter-broker requests — goes through code in this module.

Directory layout

clients/src/main/java/org/apache/kafka/
├── clients/
│   ├── ApiVersions.java, KafkaClient.java, NetworkClient.java, Metadata.java   ← shared client plumbing
│   ├── admin/                                ← KafkaAdminClient + request impls
│   ├── consumer/
│   │   ├── KafkaConsumer.java, KafkaShareConsumer.java
│   │   ├── ConsumerConfig.java, RangeAssignor.java, StickyAssignor.java, ...
│   │   └── internals/                        ← Fetcher, MembershipManager, ConsumerNetworkThread, ...
│   └── producer/
│       ├── KafkaProducer.java, ProducerConfig.java
│       └── internals/                        ← Sender, RecordAccumulator, TransactionManager
└── common/                                   ← shared types used by clients and broker
    ├── protocol/   ApiKeys, Errors, ApiMessageType, types/
    ├── network/    Selector, NetworkSend, KafkaChannel
    ├── requests/   AbstractRequest/Response wrappers
    ├── record/     RecordBatch, MemoryRecords (binary log records)
    ├── security/   SASL, SSL, scram, oauthbearer, plain, ...
    ├── serialization/  Serializer/Deserializer impls
    ├── metrics/    Sensor, Metrics, KafkaMetricsContext
    ├── compress/   Snappy/LZ4/ZSTD/GZIP wrappers
    └── utils/      Time, Utils, Bytes, KafkaThread, ...

The common/ package is shared between client and broker — core/ and the various *-coordinator modules pull it in for the wire-protocol types and shared utility code.

Key abstractions

Type	File	Purpose
KafkaProducer<K,V>	clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java	The producer entry point. Owns a RecordAccumulator and a Sender thread.
Sender, RecordAccumulator, TransactionManager	clients/src/main/java/org/apache/kafka/clients/producer/internals/	Background batching, network IO, transactional state machine.
KafkaConsumer<K,V>	clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java	The consumer entry point; supports both legacy and KIP-848 group protocols.
KafkaShareConsumer<K,V>	clients/src/main/java/org/apache/kafka/clients/consumer/KafkaShareConsumer.java	Queue-style consumer (KIP-932 share groups).
ConsumerNetworkThread, Fetcher, MembershipManager	clients/src/main/java/org/apache/kafka/clients/consumer/internals/	The new consumer's IO loop, fetch path, group membership state machine.
Admin / KafkaAdminClient	clients/src/main/java/org/apache/kafka/clients/admin/	Asynchronous admin API (createTopics, describeCluster, ACL/quota management, ...).
NetworkClient	clients/src/main/java/org/apache/kafka/clients/NetworkClient.java	Generic node-aware client used by every other client; multiplexes requests over Selector.
Metadata / MetadataSnapshot	clients/src/main/java/org/apache/kafka/clients/Metadata.java	Client-side cluster metadata cache and refresh protocol.
ApiKeys / ApiMessageType / Errors	clients/src/main/java/org/apache/kafka/common/protocol/	Numeric constants and helpers for every RPC; updated by the schema generator.
Selector, KafkaChannel	clients/src/main/java/org/apache/kafka/common/network/	Non-blocking NIO layer used by both client and broker.
RecordBatch, MemoryRecords	clients/src/main/java/org/apache/kafka/common/record/	The binary log-record format that travels on the wire and is written to disk.

Producer data flow

sequenceDiagram
    participant App
    participant Acc as RecordAccumulator
    participant Snd as Sender (thread)
    participant Net as NetworkClient
    participant Brk as Broker

    App->>Acc: send(ProducerRecord) — append to per-partition Deque
    Snd->>Acc: drain ready batches
    Snd->>Net: ProduceRequest(node, batch[])
    Net->>Brk: bytes
    Brk-->>Net: ProduceResponse
    Net-->>Snd: response
    Snd->>App: complete RecordMetadata futures, fire callbacks

Key files: KafkaProducer.java orchestrates this, but the real work is in internals/Sender.java and internals/RecordAccumulator.java. Idempotency and transactions are state-machined in internals/TransactionManager.java. See features/exactly-once.md for the protocol-level view.

Consumer data flow (new protocol)

graph LR
    App[App thread] -->|poll| EvtQ[Event queue]
    EvtQ --> CNT[ConsumerNetworkThread]
    CNT --> MM[MembershipManager]
    CNT --> Off[OffsetsRequestManager]
    CNT --> Hb[HeartbeatRequestManager]
    CNT --> Fch[FetchRequestManager]
    Fch --> Net[NetworkClient]
    Hb --> Net
    Off --> Net
    MM --> Net
    Net -->|Heartbeat / Fetch / OffsetCommit| Broker

The legacy protocol uses a similar shape but a separate heartbeat thread. Both implementations live in clients/.../consumer/internals/. The broker side of the new protocol is in group-coordinator/.

NetworkClient and the wire protocol

NetworkClient is the per-connection multiplexer used by every other client (KafkaProducer, KafkaConsumer, KafkaAdminClient, the broker's inter-broker client, KafkaRaftClient's KafkaNetworkChannel). It maintains:

• ClusterConnectionStates — per-node connection state machine (connecting, ready, disconnected, throttled).
• InFlightRequests — pending requests per node, keyed by correlation ID.
• Metadata — the cached cluster topology, refreshed on-demand via MetadataRequest.

Every request goes through ApiKeys (the master enum) and is assembled / parsed by classes generated from JSON schemas. To add a new RPC: add a JSON schema under clients/src/main/resources/common/message/, update ApiMessageType.json, and the generator does the rest.

Configuration

Each client carries its own typed config:

• ProducerConfig — clients/src/main/java/org/apache/kafka/clients/producer/ProducerConfig.java (~1,500 declared options including transactional).
• ConsumerConfig — clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java.
• AdminClientConfig — clients/src/main/java/org/apache/kafka/clients/admin/AdminClientConfig.java.
• CommonClientConfigs — shared (bootstrap servers, security, metric reporters, ...).

All three extend AbstractConfig (clients/src/main/java/org/apache/kafka/common/config/AbstractConfig.java), which normalizes raw Map<String,Object> properties against a ConfigDef schema. Adding/removing/renaming a config requires a KIP.

Security

clients/src/main/java/org/apache/kafka/common/security/ carries every supported authentication/encryption mechanism:

Mechanism	Where
TLS (PLAINTEXT/SSL)	common/security/ssl/SslEngineFactory.java and SslChannelBuilder.java
SASL/PLAIN	common/security/plain/
SASL/SCRAM (256 / 512)	common/security/scram/
SASL/OAUTHBEARER	common/security/oauthbearer/
Kerberos (GSSAPI)	common/security/kerberos/
Delegation tokens	common/security/token/
ACLs	common/acl/

The SaslChannelBuilder chooses a SASL mechanism per listener; mechanism implementations are loaded via the standard Java SASL provider mechanism plus Kafka-specific config classes.

Telemetry (KIP-714)

clients/src/main/java/org/apache/kafka/common/telemetry/ implements the broker-driven client metrics protocol, where the broker can ask clients to push selected metrics back to it. The producer/consumer/admin all delegate to a ClientTelemetryReporter.

How clients is consumed elsewhere

• core/ depends on clients for the wire-protocol types and NetworkClient. Inter-broker requests (Fetch, AlterPartition, Heartbeat between coordinators, etc.) reuse the same client classes.
• streams/ is built on top of KafkaProducer + KafkaConsumer; nothing in Streams talks to brokers except via these.
• connect/runtime/ is also a heavy clients user — Worker owns multiple producers and consumers per connector.
• tools/ uses Admin for almost every CLI tool.

Entry points for modification

• Wire-protocol changes: edit a JSON schema under clients/src/main/resources/common/message/ and run ./gradlew processMessages. Almost always needs a KIP.
• Producer behavior: start in KafkaProducer.java, then look at internals/Sender.java and internals/RecordAccumulator.java.
• Consumer rebalance / fetch: KafkaConsumer.java is the public entry; the actual logic is in internals/. The new (KIP-848) implementation has a richer MembershipManager state machine.
• Adding a SASL mechanism: a new package under common/security/ plus updates to SaslChannelBuilder. KIP required.

Related pages

• Modules: core (broker) — broker side of the protocol.
• Modules: tools — CLI commands built on clients.
• Features: Exactly-once semantics.
• Features: ACLs and security.