README

lisachenko/kafka-client is a native, pure-PHP implementation of the Apache Kafka wire protocol — no ext-rdkafka required. It ships a Producer, a Consumer and a low-level Admin client, designed to stay close in spirit to the official Java client's API while feeling natural in PHP.

This branch is the 4.x line and speaks the Apache Kafka 4.3.1 wire protocol — everything Kafka 4.0 to 4.3 added to the apis the client implements, the transaction protocol v2 of KIP-890 part 2, Produce and the group offsets by topic id among them, and the KIP-932 share consumer (Consumer\KafkaShareConsumer) with its admin methods — on top of the KIP-848 consumer protocol of the finished 3.x line (branched off as 3.x), built one Kafka minor at a time and verified against a 4.3.1 KRaft node; the versions Kafka 4.0 removed (KIP-896) are never sent to a 4.x node. The version of every api the client sends is listed under Supported Kafka protocol versions. main is the top of the cascade: the frozen protocol snapshots below it live on 3.x (Kafka 3.9.2), 2.x (Kafka 2.8.2), 1.x (Kafka 1.1.1), 0.11.x (Kafka 0.11.0.3), 0.10.x (Kafka 0.10.2.2), 0.9.x (Kafka 0.9.0.1) and 0.8.x (Kafka 0.8.2.2), and every wire vector those lines captured is replayed against the classes of this branch. The grammar this branch implements is written down, byte for byte, in docs/protocol/4.3.md, verified against a real Kafka 4.3.1 node running in KRaft mode (the frames of the 3.x line were captured on a 3.9.2 one); the record of the 4.x line — its release notes above the plan it was built from — is docs/handoff/main.md, and the records of the finished lines are docs/handoff/3.x.md, docs/handoff/2.x.md and docs/handoff/1.x.md.

Installation

composer require lisachenko/kafka-client:dev-main

PHP 8.4 or newer, and nothing else — ext-openssl is needed only for SSL/SASL_SSL and ext-zlib (bundled with PHP) for gzip and ext-zstd for the zstd codec of Kafka 2.1; the snappy and lz4 codecs are implemented in PHP and use ext-snappy only when it happens to be installed. main is the branch the top of the cascade lives on, so it is installed by branch name; the frozen lines below it carry a numeric branch and are installed by constraint (^2.8@dev for 2.x, ^1.1@dev for 1.x, ^0.11@dev for 0.11.x, ^0.10@dev for 0.10.x, and so on). A line is frozen as a numeric branch when the line above it starts, so code that must keep speaking Kafka 2.8.2 pins the branch rather than dev-main.

Producer API

The Producer API sends streams of records to topics in the Kafka cluster.

use Protocol\Kafka\Common\Record\Record;
use Protocol\Kafka\Producer\KafkaProducer;
use Protocol\Kafka\Producer\ProducerConfig;
use Protocol\Kafka\Producer\RecordMetadata;

require __DIR__ . '/vendor/autoload.php';

$producer = new KafkaProducer([
    ProducerConfig::BOOTSTRAP_SERVERS => ['tcp://127.0.0.1:9092'],
    ProducerConfig::ACKS              => 1,
]);

$producer->send('test', new Record('foo'))->then(
    function (RecordMetadata $metadata): void {
        echo "Written to partition {$metadata->partition} at offset {$metadata->offset}\n";
        echo "The log holds the timestamp {$metadata->timestamp}\n";
        echo "The broker throttled the batch for {$metadata->throttleTimeMs} ms\n";
    }
);
$producer->flush();

send(string $topic, Record $record, ?int $partition = null): Promise buffers the record and returns a promise that is resolved with a RecordMetadata once flush() has sent the batch and the broker has acknowledged it. The only required option is ProducerConfig::BOOTSTRAP_SERVERS; for every other option see the constants documented on Protocol\Kafka\Producer\ProducerConfig and the producer configuration reference.

ProducerConfig::ACKS selects the durability of a write: 0 sends fire-and-forget (the broker sends no response at all for such a request, so the promise resolves with the offset -1), 1 waits for the leader's log and -1 for all in-sync replicas. Records are collected until they fill ProducerConfig::BATCH_SIZE bytes or ProducerConfig::LINGER_MS has passed, and a batch that fails with a retriable error is sent again ProducerConfig::RETRIES times — that option is the whole retry budget of a batch and defaults to no retry at all, like the Java producer. Compression is set with ProducerConfig::COMPRESSION_TYPE and applies to a whole batch: gzip, snappy, lz4 (Kafka 0.10.0) and zstd (Kafka 2.1, KIP-110, through ext-zstd), in the frame format of the Kafka producer including the KAFKA-3160 checksum quirk of a message format v0 frame.

Message formats, timestamps and headers. Kafka 0.10.0 gave every record a timestamp: Record::$timestamp (milliseconds since the epoch) and Record::$timestampType (TimestampType::CREATE_TIME, LOG_APPEND_TIME or NO_TIMESTAMP_TYPE), and Kafka 0.11 gave it headers (Record::withHeaders(), Common\Record\Header), a list of key-value pairs of metadata next to the key and the value. send() stamps the create time of every record that does not carry one, and ProducerConfig::MESSAGE_FORMAT_VERSION (message.format.version, 0.11.0 by default) selects the format a batch is written in — the record batch v2 by default, 0.10.x for a message set with timestamps and 0.9.0 for one without. The format decides the version of the Produce request: only the message format v2 travels in a Produce v12, and only it has a place for the headers, for the producer id of an idempotent producer and for a transaction; a message set is sent as a Produce v2, and a 2.8.2 broker answers 87 INVALID_RECORD for every partition of a Produce v3 or above that carries one. RecordMetadata::$timestamp reports what the log holds: the create time of the first record of the batch, or the LogAppendTime the broker answered with (Produce v2 and above) when the topic is configured with message.timestamp.type=LogAppendTime. Version 5 (Kafka 1.0) also reports the logStartOffset of every partition it answers — the first offset the log still holds after a retention run or a deleteRecords() — on ProduceResponsePartition.

Without a key a record is spread over the partitions that have a leader, with a key it goes to the partition that the murmur2 hash of the key selects, exactly as with the official Java client (Producer\DefaultPartitioner); an explicit partition can be passed to send().

RecordMetadata::$throttleTimeMs is the ThrottleTime that version 1 of the Produce API added in Kafka 0.9: the number of milliseconds the broker delayed the answer of that batch because the client.id exceeded its producer_byte_rate quota. Quotas never reject a write — the records are appended and only the response is held back — so the field is informational, and it is 0 on a broker without quotas as well as for a fire-and-forget batch (ACKS => 0), which is never answered. The consumer side is the same: every Common\FetchedPartition of Client::fetchPartitions() carries the throttleTimeMs of the Fetch answer it came in. Quotas are set per client id on a running broker, e.g.

$ kafka-configs.sh --bootstrap-server localhost:9092 --alter \
    --add-config 'producer_byte_rate=1024,consumer_byte_rate=2048' \
    --entity-type clients --entity-name my-application

A runnable version of this is examples/producer.php.

Idempotent producer

Kafka 0.11 added a delivery guarantee that no release before it had, and one option turns it on:

$producer = new KafkaProducer([
    ProducerConfig::BOOTSTRAP_SERVERS  => ['tcp://127.0.0.1:9092'],
    ProducerConfig::ENABLE_IDEMPOTENCE => true,
]);

With enable.idempotence the producer asks a broker for a producer id before its first batch (InitProducerId, key 22) and numbers the batch of every topic-partition with a gapless sequence number. A batch that has to be sent again — a lost acknowledgement, a leader that moved — goes out with the very same producer id, epoch and sequence numbers, and the broker recognises it as the batch it already holds: it answers the offset of the original append and writes nothing. The records of a partition therefore reach the log exactly once and in order, however often the client had to retry, and nothing about the API changes: send() and flush() work as before.

The guarantee implies acks = all and a non-zero retries; both are set for you when you did not set them (retries becomes 3, where the Java producer, which has a background sender, uses an unbounded budget), and a configuration that contradicts them — acks of 0 or 1, or retries of 0 — is refused with an InvalidConfigurationException. The Java requirement of max.in.flight.requests.per.connection = 1 needs no option here: this client sends one produce request at a time.

It holds within one producer session: a new KafkaProducer gets a new producer id and cannot deduplicate against what the previous one wrote, and a record your application sends a second time is a new batch, which the broker has no way of recognising. Deduplication across sessions is what a transactional.id is for.

A Kafka 1.x broker widens the guarantee in two ways that need no option: it recognises a duplicate of any of the last five batches of a producer and partition, not only of the very last one, so a producer whose acknowledgements of several batches in a row were lost is still answered with the original offsets instead of being thrown out of sequence; and it tells a client when it has lost the state of a producer altogether, which a 0.11 broker could not.

Three error codes of the broker say something about the producer state itself. 47 (ProducerFencedException) means another producer took the producer id over: the producer is finished and refuses every further send. 45 (OutOfOrderSequenceException) means the producer and the broker no longer agree on what is in the log: the batch that hit it is reported to the caller, and the producer starts over with a new producer id — everything written under the old one loses its deduplication. 59 (UnknownProducerIdException, Kafka 1.0, a subclass of the previous one) means the broker has no state of this producer for that partition — because every record it wrote there was deleted by deleteRecords() or by a retention run. That one the producer repairs by itself: the logStartOffset that Produce v5 added to the answer shows that the records fell below the start of the log, so the partition is numbered from the sequence 0 again and the batch is sent once more, under the same producer id and without touching any other partition. All three are documented, with what a real 1.1.1 broker answers, in docs/protocol/4.3.md, section "The idempotent producer".

Transactions

A transactional.id turns the idempotent producer into a transactional one: the records of several partitions — and the committed offsets of a consumer group — become one unit that a read_committed consumer either sees whole or does not see at all, and the guarantee survives a restart of the producer, because the id is what the broker remembers it by.

use Protocol\Kafka\Common\Errors\KafkaException;
use Protocol\Kafka\Common\Record\Record;
use Protocol\Kafka\Producer\KafkaProducer;
use Protocol\Kafka\Producer\ProducerConfig;

$producer = new KafkaProducer([
    ProducerConfig::BOOTSTRAP_SERVERS => ['tcp://127.0.0.1:9092'],
    ProducerConfig::TRANSACTIONAL_ID  => 'orders-etl-1',   // implies enable.idempotence
]);

$producer->initTransactions();          // once, before the first send

$producer->beginTransaction();
try {
    $producer->send('orders', Record::fromValue('one'));
    $producer->send('audit',  Record::fromValue('one accepted'));
    $producer->commitTransaction();     // flushes what is buffered, then EndTxn
} catch (KafkaException $error) {
    $producer->abortTransaction();      // the only way out of a failed transaction
}

initTransactions() asks the transaction coordinator of the id for a producer id and an epoch one higher than the previous incarnation used, which fences that incarnation for good and rolls back whatever transaction it left open — so a transactional id must be used by one producer at a time, and a crashed producer never blocks a reader for longer than its transaction.timeout.ms. A send() outside a transaction is refused, and so is one after an error that only an abort can clean up; commitTransaction() flushes the buffer before it ends the transaction and abortTransaction() throws it away.

The read side is one consumer option:

$consumer = new KafkaConsumer([
    ConsumerConfig::BOOTSTRAP_SERVERS  => ['tcp://127.0.0.1:9092'],
    ConsumerConfig::GROUP_ID           => 'orders-readers',
    ConsumerConfig::ISOLATION_LEVEL    => ConsumerConfig::ISOLATION_LEVEL_READ_COMMITTED,
]);

With read_committed the broker answers only up to the last stable offset of a partition, so nothing of a transaction that is still open is shown, endOffsets() reports the offset such a reader can really reach, and the records of transactions the broker names as aborted are dropped by the consumer before poll() returns — the broker sends them and only names them. The COMMIT and ABORT control batches of a transaction never reach an application in either level.

The consume-transform-produce loop is what all of this exists for: the consumer hands its offsets to the producer instead of committing them itself, so reading the input and writing the output either both happen or neither does.

$producer->beginTransaction();
foreach ($consumer->poll(1000)['input'][0] ?? [] as $record) {
    $producer->send('output', Record::fromValue(strtoupper((string) $record->value)));
}
$producer->flush();
$producer->sendOffsetsToTransaction(['input' => [0 => $consumer->position('input', 0)]], 'my-group');
$producer->commitTransaction();

The consumer of that loop runs with enable.auto.commit = false and read_committed. A runnable version is examples/transactional-producer.php; the wire protocol behind it — the five apis 24 to 28, the control batches and the last stable offset — is in docs/protocol/4.3.md, section "Transactions".

Consumer API

The Consumer API reads streams of records from topics in the Kafka cluster. Kafka 0.9 moved the coordination of a consumer group into the broker, so a consumer can simply subscribe to topics and let the group hand out the partitions:

use Protocol\Kafka\Common\ClientConfig;
use Protocol\Kafka\Consumer\ConsumerConfig;
use Protocol\Kafka\Consumer\KafkaConsumer;
use Protocol\Kafka\Consumer\OffsetResetStrategy;

$consumer = new KafkaConsumer([
    ClientConfig::BOOTSTRAP_SERVERS => ['tcp://127.0.0.1:9092'],
    // A JoinGroup is answered only once the whole rebalance is over, so this has to exceed both
    // session.timeout.ms and max.poll.interval.ms; the consumer defaults are those of the Java
    // consumer of 0.10.1: 305000, 10000 and 300000
    ClientConfig::REQUEST_TIMEOUT_MS => 305000,

    ConsumerConfig::GROUP_ID                      => 'kafka-daemon',
    ConsumerConfig::PARTITION_ASSIGNMENT_STRATEGY => 'range', // or 'roundrobin', or your own class
    ConsumerConfig::SESSION_TIMEOUT_MS            => 10000,
    ConsumerConfig::MAX_POLL_INTERVAL_MS          => 300000,
    ConsumerConfig::HEARTBEAT_INTERVAL_MS         => 3000,
    ConsumerConfig::AUTO_OFFSET_RESET             => OffsetResetStrategy::EARLIEST,
]);

// Nothing is sent yet: the group is joined by the first poll(), which brings the assignment
$consumer->subscribe(['test']);

while (true) {
    // [topic][partition] => records, in offset order
    foreach ($consumer->poll(1000) as $topic => $partitions) {
        foreach ($partitions as $partition => $records) {
            foreach ($records as $record) {
                // $record->timestamp and $record->timestampType come from message format v1
                echo $topic, ':', $partition, '@', $record->offset, ' ', $record->value, PHP_EOL;
            }
        }
    }
    $consumer->commitSync();
}

$consumer->close(); // commits once more and leaves the group with a LeaveGroup request

subscribe() names the topics and assignment() reports the partitions the group gave this member; assign() still picks partitions by hand and joins no group at all, and the two are mutually exclusive, exactly as in the Java client. commitSync() stores the position of the group — carrying the member id and the generation of this consumer, so a coordinator refuses a commit of a generation that is over — and seek()/seekToBeginning()/seekToEnd() move the position. unsubscribe() leaves the group without committing, close() commits first.

The heartbeat is sent from poll(), because PHP has no background thread. A consumer that does not poll for longer than session.timeout.ms is dropped by the coordinator and its partitions are given to the other members; the next poll() sees that in the error code of its heartbeat and joins the group again. Keep the processing of a batch well below the session timeout, or raise session.timeout.ms — within the group.min.session.timeout.ms and group.max.session.timeout.ms of the broker.

max.poll.interval.ms (Kafka 0.10.1, KIP-62) is the second half of that: it is sent as the rebalance_timeout of the JoinGroup v1 request and tells the coordinator how long it should wait for this member to rejoin a rebalance — the group waits that long instead of the session timeout, which is what lets a member process a batch for minutes without holding up nothing but its own rejoin. In a PHP consumer it does not evict anything the way it does in a Java one: without a heartbeat thread, an application that stops polling stops heartbeating and is dropped when its session timeout expires. request.timeout.ms has to exceed both timeouts, because a JoinGroup blocks the connection for the whole rebalance, and subscribe() refuses a value that does not.

Record headers and the isolation level are what Kafka 0.11 adds on this side. A record read out of a record batch v2 carries the headers the producer wrote (ConsumerRecord::$headers, a list of Common\Record\Header), next to the key, the value, the timestamp and its type; a topic whose message.format.version is older simply has none. ConsumerConfig::ISOLATION_LEVEL (isolation.level, read_uncommitted by default) is sent as the isolation level of the Fetch v16 request: with read_committed the broker answers only up to the last stable offset — the first record of a transaction that has neither committed nor aborted — and names the aborted transactions of the answer, whose records the consumer drops. The control batches of the transaction protocol are never handed to an application in either level. The option travels in the Offsets v6 request as well, so endOffsets(), position() and seekToEnd() of a read_committed consumer answer the last stable offset instead of the log end offset — a consumer that compares its position against the end of a partition compares it against the offset it can really reach. AdminClient::listOffsets() stays at read_uncommitted on purpose: an administrator asks what is in the log.

Incremental fetch sessions (Kafka 1.1, KIP-227) are what the consumer adds on this side. Every broker it reads from holds a fetch session for it: the first request states the whole assignment and opens the session, every following one states only the partitions whose position moved and lets the broker fill in the rest, a partition that leaves the assignment — a rebalance, pause(), a topic that is gone — is dropped from the session with the forgotten_topics_data of the next request, and the answer carries only the partitions that have news. A consumer of many partitions therefore stops repeating its partition list in every fetch, and the broker stops answering partitions that have nothing to say. None of it is visible in poll(): the error codes 70 (FetchSessionIdNotFound, the broker no longer knows the session) and 71 (InvalidFetchSessionEpoch, a request or an answer was lost) are answered with a full fetch by the client itself, in the same call, and a broker that hands out no session at all — one below Kafka 1.1, or one whose cache of 1000 sessions is full — leaves the consumer on plain full fetches. Client::fetchPartitions() keeps its session-less behaviour for callers that want one request and one answer; the consumer fetches through Client::fetchPartitionsWithSessions().

Offsets by timestamp (Kafka 0.10.1, KIP-79) are what the record timestamps buy on the consumer side: offsetsForTimes(['test' => [0 => $millis]]) answers the first record of each partition whose timestamp is at or after the given one, as an OffsetAndTimestamp (or null when the partition holds no such record), and beginningOffsets() / endOffsets() are the two special timestamps -2 and -1. All three are pure queries and need no assignment, exactly as in the Java consumer.

The partitions are distributed by the member the coordinator elected as the leader of the generation: partition.assignment.strategy selects range (the default) or roundrobin — both with the ordering rules of the Java client of 0.9.0.1, so a PHP member can lead a group of Java members and the other way round — or names a class that implements Consumer\PartitionAssignorInterface. Every member of a group has to offer the same one, a coordinator that finds no common protocol refuses the join with the error 23. A Consumer\ConsumerRebalanceListener passed to subscribe() is called with the partitions that each rebalance takes away and hands over, which is where a consumer with enable.auto.commit off commits what it has consumed.

group.protocol = consumer switches the consumer to the new consumer protocol of KIP-848 (Kafka 3.5). The four apis of the classic membership protocol are then replaced by the single ConsumerGroupHeartbeat (key 68), and three things change for an application: the coordinator computes the assignment, so partition.assignment.strategy has nothing to say and group.remote.assignor names a server-side assignor instead; the heartbeat interval is dictated by the broker (group.consumer.heartbeat.interval.ms) rather than by heartbeat.interval.ms; and a rebalance is incremental — a member gives up only the partitions it really loses, so onPartitionsRevoked() sees exactly those and onPartitionsAssigned() only the ones that were added, where the classic protocol hands the whole assignment back and forth on every rebalance. The member epoch takes the place of the generation and travels in the OffsetCommit v10 and OffsetFetch v10 of that member (v9 and v9 below Kafka 4.2). A group is of one protocol or the other: a heartbeat for a classic group is refused with the 69, and AdminClient::describeConsumerGroups() (key 69) describes the new groups where describeGroups() (key 15) describes the classic ones.

examples/consumer-group.php is a runnable version of this — start it twice and watch the two members split the partitions — and examples/consumer.php is the same thing with assign().

See the consumer configuration reference for the full set of options.

Share consumer API

A share group (KIP-932, Kafka 4.1) is a group whose members do not own partitions but share them record by record: every record of a subscribed topic is delivered to one member at a time, which holds it under an acquisition lock (the group config share.record.lock.duration.ms, 30 s by default) and acknowledges it. Consumer\KafkaShareConsumer is the KafkaShareConsumer of the Java client @ 4.3.1:

use Protocol\Kafka\Consumer\AcknowledgeType;
use Protocol\Kafka\Consumer\ConsumerConfig;
use Protocol\Kafka\Consumer\KafkaShareConsumer;

$consumer = new KafkaShareConsumer([
    ConsumerConfig::BOOTSTRAP_SERVERS          => ['tcp://127.0.0.1:9092'],
    ConsumerConfig::GROUP_ID                   => 'my-share-group',
    ConsumerConfig::SHARE_ACKNOWLEDGEMENT_MODE => ConsumerConfig::SHARE_ACKNOWLEDGEMENT_MODE_EXPLICIT,
]);
$consumer->setAcknowledgementCommitCallback(function (array $offsets, $exception) {
    // [topic][partition] => offsets, and the error of the node or null - once per topic-partition
});
$consumer->subscribe(['my-topic']);

while (true) {
    foreach ($consumer->poll(1000) as $topic => $partitions) {        // joins the group on the first call
        foreach ($partitions as $partition => $records) {
            foreach ($records as $record) {                          // ConsumerRecord, with $record->deliveryCount
                $consumer->acknowledge($record, process($record) ? AcknowledgeType::ACCEPT : AcknowledgeType::RELEASE);
            }
        }
    }
    $results = $consumer->commitSync();                              // [topic][partition] => ?KafkaException
}
$consumer->close();                  // closes the share sessions (the rest is released) and leaves the group

An acknowledgement is one of four: ACCEPT (processed, never delivered again), RELEASE (delivered again - to this member or another one - with a delivery count one higher, until the group config share.delivery.count.limit archives it), REJECT (archived at once) and RENEW (Kafka 4.2, KIP-1222: still being processed - the lock starts over, and the next poll() returns the record again). share.acknowledgement.mode is implicit by default: the next poll() or commit then accepts everything the last poll() returned, and acknowledge() is refused; in the explicit mode every record has to be acknowledged before the next poll(). Acknowledgements travel with the ShareFetch of the next poll(), or in a ShareAcknowledge of commitSync()/commitAsync(); acknowledge($topic, $partition, $offset, $type) is the form for a record a deserializer failed on. share.acquire.mode = record_limit (KIP-1206) has the node acquire no more than max.poll.records records per fetch, where the default batch_optimized hands out whole record batches, and acquisitionLockTimeoutMs() reports the lock of the last answer.

Where to start and what to show are group configs, not options of the consumer: a share group reads from latest unless its share.auto.offset.reset says earliest, and its isolation level is share.isolation.level - set them with IncrementalAlterConfigs on the config resource type 32 (ConfigResource.Type.GROUP) before the first member joins. The constructor refuses auto.offset.reset, enable.auto.commit, group.instance.id, isolation.level, partition.assignment.strategy, session.timeout.ms, heartbeat.interval.ms, group.protocol and group.remote.assignor, as the Java one does.

PHP has no background thread, so the heartbeat of a share member is sent from poll() (at the interval the coordinator dictates), commitAsync() sends its request before it returns and reports only to the callback, and the callback runs inside the call that sent the acknowledgements. A member keeps one share session per leader: the ShareFetch of the epoch 0 opens it, and a session the node lost - it lives on its connection, so a dropped connection takes it along - is answered 122 or 123, after which the consumer opens a new one by itself; the acknowledgements of the lost session are reported as failed to the callback, and its records are delivered again. AdminClient::describeShareGroups() describes the group, and deleteConsumerGroups() deletes it once it is empty.

examples/share-consumer.php is a runnable version of this — start it twice and watch two members share one topic. The protocol document has the frames and what the node answered in "The share consumer (KIP-932)".

Admin API

The Admin API exposes the low-level cluster operations a 1.1.1 broker can serve:

use Protocol\Kafka\Admin\AdminClient;
use Protocol\Kafka\Admin\ConfigResource;
use Protocol\Kafka\Admin\NewTopic;
use Protocol\Kafka\Admin\RecordsToDelete;
use Protocol\Kafka\Common\ClientConfig;
use Protocol\Kafka\Common\Cluster;
use Protocol\Kafka\Protocol\Request\OffsetsRequest;

$configuration = [ClientConfig::BOOTSTRAP_SERVERS => ['tcp://127.0.0.1:9092']];
$admin         = new AdminClient(Cluster::bootstrap($configuration), $configuration);

$brokers  = $admin->findAllBrokers();                       // Node[], indexed by the node id
$apis     = $admin->getApiVersions($brokers[0]);            // api key => ApiVersionsResponseMetadata
$topics   = $admin->listTopics();                           // string[]
$metadata = $admin->describeTopics(['test']);               // TopicMetadata[], indexed by the topic
$offsets  = $admin->listOffsets(['test' => [0, 1, 2]]);     // topic => partition => offset
$earliest = $admin->listOffsets(['test' => [0]], OffsetsRequest::EARLIEST);

$controller = $admin->findController();                     // Node, from the controller_id of Metadata v1
$created    = $admin->createTopics([new NewTopic('test-2', 3, 1)]);   // topic => ?KafkaException
$deleted    = $admin->deleteTopics(['test-2']);                      // topic => ?KafkaException

$purged = $admin->deleteRecords(['test' => [0 => 100]]);    // topic => partition => DeletedRecords (low watermark)
$purged = $admin->deleteRecords(['test' => [0 => RecordsToDelete::allRecords()]]);

$topicResource = ConfigResource::topic('test');
$configs       = $admin->describeConfigs([$topicResource]); // resource key => Config
echo $configs[$topicResource->key()]->value('retention.ms');
$altered = $admin->alterConfigs([                           // resource key => ?KafkaException
    $topicResource->key() => ['retention.ms' => '3600000'] + $configs[$topicResource->key()]->nonDefaultValues(),
]);

$coordinator = $admin->findCoordinator('kafka-daemon');     // Node that holds the group offsets
$committed   = $admin->listGroupOffsets('kafka-daemon');    // every topic the group committed (v2)

$groups = $admin->listAllGroups();                          // group id => ListGroupResponseProtocol
$groups = $admin->listGroups($coordinator);                 // only the groups of that one broker

$group = $admin->describeGroup('kafka-daemon');             // DescribeGroupResponseMetadata
echo $group->state;                                         // Stable, CompletingRebalance, PreparingRebalance, Empty or Dead
echo $group->protocol;                                      // the assignor, only while the group is stable
foreach ($group->members as $memberId => $member) {
    echo $memberId, ' ', $member->clientId, ' ', $member->clientHost, PHP_EOL;
    // $member->memberMetadata and $member->memberAssignment are the opaque bytes of the protocol type
}

The three topic apis — createTopics(), deleteTopics() and createPartitions() — are served by the controller alone: AdminClient looks it up in the controller_id of a Metadata answer, and repeats the request once against a freshly looked up controller when a topic comes back with the error code 41 (NotController). None of them throws for a topic: the result has one entry per requested topic, in the order of the request, null when it worked and the exception of its error code — with the error_message the controller sent in the context — when it did not, because one topic of a batch says nothing about the others. deleteConsumerGroups() reports its groups the same way, and sends one request to the coordinator of each of them.

getApiVersions() is what Kafka 0.10.0 added: it asks one broker for the version range of every api it serves and returns them indexed by the api key, which is the only way to tell one release of the protocol from another without guessing. Kafka 0.11 raised it to version 1, whose answer carries a trailing throttle_time_ms — the one api of KIP-124 that appends the field instead of prepending it, because an unknown version is still answered in the version 0 layout. Every broker answers for itself, so a rolling upgrade shows up as brokers that report different ranges. Client::apiVersions() returns the whole response, with supports() and maxVersionOf() on it.

The group apis are what Kafka 0.9 added when it moved the consumer groups out of ZooKeeper, and Kafka 0.10.1 gave them one more state: a group exists on its coordinator from the first JoinGroup until its committed offsets expire, so listGroups() shows it even after its last member has left, and describeGroup() reports its state, the assignor its members agreed on and one entry per member, with the Subscription and MemberAssignment of the consumer protocol as opaque byte arrays. A group with no members left is Empty, not Dead; asking about a group that does not exist is still not an error, the coordinator answers the state Dead with the error code 0.

Creating a topic no longer means writing to ZooKeeper: CreateTopics (key 19) and DeleteTopics (key 20) arrived with Kafka 0.10.1, and AdminClient::createTopics() sends the version 7 of the first one, with validate_only, the per-topic error_message, the shape and configuration of the new topic in the answer (KIP-525) and the topic id of KIP-516 next to them. The implicit creation by a Metadata request of an unknown topic still works when the broker runs with auto.create.topics.enable=true, and still answers the topic error code 5 (LeaderNotAvailable) with an empty partition list until the controller has elected the leaders — but the admin client no longer triggers it: Metadata v4 (Kafka 0.11, KIP-4) added allow_auto_topic_creation, and every request of AdminClient sends it as false, so describing a topic that does not exist is answered with the code 3 and creates nothing. createTopics() is the explicit alternative that reports what went wrong. Metadata v5 (Kafka 1.0, KIP-112/113) is what this client sends today, so every partition it describes also carries its offlineReplicas — the replicas whose broker is down or whose log directory failed.

Metadata v1 and v2 also gave the cluster an identity of its own: Cluster::clusterId() is the cluster_id the broker generated (the /cluster/id znode), Cluster::controller() the node the controller_id names, Common\Node::$rack the broker.rack of a broker, and Cluster::topics() hides __consumer_offsets unless exclude.internal.topics is turned off.

Records and configuration through the protocol are what Kafka 0.11 added to the admin surface. deleteRecords() (KIP-107) moves the low watermark of a partition forward — everything below the offset becomes unreadable at once, the record at the offset stays — and answers the new watermark of every partition as an Admin\DeletedRecords; the offset is a plain integer or an Admin\RecordsToDelete (beforeOffset(), or allRecords() for the -1 of the wire, i.e. up to the high watermark). It is served by the leader of each partition, so the request is split per leader and a partial failure is reported as a TopicPartitionRequestException. describeConfigs() and alterConfigs() (KIP-133) read and write the configuration of a topic or of a broker without going through ZooKeeper: a resource is an Admin\ConfigResource (topic() / broker()) and is addressed in the result by its key(), because PHP cannot use an object as an array key. alterConfigs() replaces the whole configuration of a topic — an option that is left out is reset to its default, which is what Config::ownValues() exists for (nonDefaultValues() is the 0.11 name and, since KIP-226, also reports options that only the broker configuration sets). A broker resource is where this line differs from the one below it: KIP-226 made a 1.1 broker accept one and validate it per option, so an option it cannot change at runtime comes back as the error code 42 with Cannot update these configs dynamically: Set(log.retention.hours) while a dynamic one is applied, where a 0.11 broker refused every broker resource outright. Reading such a resource changed too — the is_default of an entry is derived from the KIP-226 config source and is_read_only means "not dynamically updatable". This client sends DescribeConfigs v1, which reports that source directly and, with $includeSynonyms, every place the broker looked for the value; the version 0 frame is kept for the vectors of the line below and derives the source back from the boolean, which is lossy.

The disks of a broker and the tokens of a principal are the two api families Kafka 1.x added on top of that. describeLogDirs() and alterReplicaLogDirs() (KIP-113) say which log.dirs entry a replica lives in and move it to another one — both broker-local, so they are addressed by broker id and by an Admin\TopicPartitionReplica rather than by a partition leader. The four token apis of KIP-48 issue, renew, expire and describe a delegation token over an authenticated connection; what they cannot do is use one, because authenticating with a token is a SASL/SCRAM login.

examples/admin.php, examples/create-topic.php, examples/admin-configs.php, examples/admin-log-dirs.php and examples/delegation-tokens.php run all of it against the broker of docker-compose.yml.

Network client

One connection per broker is opened on demand and kept open for the requests that follow, the way a Kafka connection is meant to be used: it is an ordered request/response channel, and every request carries a correlation id that the broker echoes back. The client generates that id, checks it on every answer and drops a connection whose answer does not match — its stream position would be unknown from then on. Protocol\Kafka\Common\Node::closeConnections() closes every connection of the process, which a long-running worker can call when it goes idle.

Three options steer this:

• connections.max.idle.ms — a cached connection that was unused for longer is re-opened instead of handed out, because the broker closes idle connections on its side and a half-closed socket would only surface mid-request.
• metadata.max.age.ms — how long the cluster metadata (and a metadata.cache.file, if configured) stays valid before it is fetched again.
• retries and retry.backoff.ms — how often a request that failed with something a metadata refresh can cure is refreshed and sent again: the error codes 3 (UnknownTopicOrPartition, e.g. a topic that was only just auto-created), 5 (LeaderNotAvailable, an election is in progress) and 6 (NotLeaderForPartition, the cached leader moved), plus a dropped connection. Every other error is final and reaches the caller straight away.

A request that fans out over several partition leaders can fail for some partitions and succeed for others. That is reported as a Common\Errors\TopicPartitionRequestException, which carries both halves: getPartialResult() holds the topic-partitions that did work and getExceptions() the exception of each one that did not, indexed by topic and partition.

The Admin API uses the same connections and the same correlation id checks; it does not retry, but every request that any broker can answer — Metadata and DescribeCluster — is tried on the brokers of the cluster in turn until one of them answers. controlledShutdown() is gone from this line: ControlledShutdown (key 7) is served on the controller listener of a KRaft node only, never on a client listener.

PHP-specific configuration

A few configuration options exist purely to make the client work well under PHP's process-per-request model:

• metadata.cache.file — file used to cache cluster metadata; effectively cached by opcache in production.
• stream.async.connect — whether to connect to brokers asynchronously.
• stream.persistent.connection — whether to keep a persistent connection to the cluster.

For publishing from web requests, enabling persistent connections together with a metadata cache file keeps producing as fast as possible.

The offsets of a consumer group always go to the coordinator of the group, with OffsetCommit v10 and OffsetFetch v10 (topics named by id, Kafka 4.2), and live in the __consumer_offsets topic: the commit carries the member id, the generation and the group.instance.id of a group member and the leader epoch of every offset, but no RetentionTime any more (KIP-211 took the field out at version 5, so offsets.retention.minutes of the broker alone decides), and the fetch is the one that can ask for every topic the group committed and the one that can insist on stable offsets (KIP-447). The ZooKeeper storage of Kafka 0.8.1 (the version 0 of both apis, the offsets.storage option of the lines below) is gone from this line: a KRaft node answers both v0 requests with 35.

Configuration reference

Every option is a plain array key of the configuration passed to KafkaProducer, KafkaConsumer or AdminClient; the constants are documented one by one on Common\ClientConfig, Consumer\ConsumerConfig and Producer\ProducerConfig. The options the 0.10 line adds are marked (0.10).

Client (Common\ClientConfig, shared by all three)

Consumer (Consumer\ConsumerConfig)

Producer (Producer\ProducerConfig)

Security / SSL

Kafka 0.9 is the release that added transport security and Kafka 0.10.0 the release that made authentication part of the protocol: a broker binds one listener per security protocol (listeners=PLAINTEXT://…,SSL://…,SASL_PLAINTEXT://…,SASL_SSL://…) and every listener answers the identical request set, so the transport changes and never a single byte of a request. Point bootstrap.servers at the listener and set security.protocol:

use Protocol\Kafka\Common\ClientConfig;
use Protocol\Kafka\Common\Security\SecurityProtocol;
use Protocol\Kafka\Producer\KafkaProducer;

$producer = new KafkaProducer([
    ClientConfig::BOOTSTRAP_SERVERS    => ['tcp://kafka-1.example.com:9093'],
    ClientConfig::SECURITY_PROTOCOL    => SecurityProtocol::SSL,
    ClientConfig::SSL_CA_CERT_LOCATION => '/etc/kafka/ca.pem',
]);

The certificate of the broker is always verified, and its subject has to match the host the connection was made to — a self-signed broker certificate therefore needs ssl.ca.cert.location pointing at it. The handshake happens right after connect() and is bounded by the connection timeout of the stream, not by request.timeout.ms.

Metadata over a listener. Every version of the Metadata api has room for exactly one host/port per broker, and the broker fills it with the endpoint of the listener the request arrived on. A client that bootstraps over TLS therefore learns the TLS endpoints of the whole cluster and keeps talking TLS to every broker it discovers; one that bootstraps in plaintext learns the plaintext ones, one that authenticates learns the SASL ones. They never mix, and there is no way to ask one listener about another.

examples/ssl.php produces and consumes over the SSL listener of docker-compose.yml, whose self-signed certificate is checked in as docker/kafka-4.3.1/ssl/broker.crt.

SASL/PLAIN works on this branch. Kafka 0.9 did have SASL, but only GSSAPI (Kerberos) and negotiated outside the Kafka protocol; Kafka 0.10.0 (KIP-43) added the SaslHandshake request (api key 17) and the PLAIN mechanism, which is what makes authentication implementable in pure PHP, and Kafka 1.0 (KIP-152) added the SaslAuthenticate request (api key 36), which is what makes a refused password reportable:

$producer = new KafkaProducer([
    ClientConfig::BOOTSTRAP_SERVERS => ['tcp://kafka-1.example.com:9094'],
    ClientConfig::SECURITY_PROTOCOL => SecurityProtocol::SASL_PLAINTEXT, // or SASL_SSL
    ClientConfig::SASL_MECHANISM    => SaslMechanism::PLAIN,
    ClientConfig::SASL_USERNAME     => 'kafkatest',
    ClientConfig::SASL_PASSWORD     => 'kafkatest-secret',
]);

The handshake and the token exchange happen inside connect(), before the first ordinary request: one SaslHandshake frame naming the mechanism, then the PLAIN token \0<username>\0<password>, answered with an empty token. Kafka 1.0 (KIP-152) gave that token a request of its own — SaslAuthenticate, api key 36 — and version 1 of the handshake is how a client asks for it; this client sends v1, so the token travels as an ordinary framed request and a refused credential comes back as the error code 58 with the message of the broker (Authentication failed: Invalid username or password) instead of a silently closed socket. The raw, unframed exchange of a v0 handshake is still implemented and still served by a 1.1.1 broker — it is what the four lines below speak. PLAIN sends the password in clear text, so use SASL_SSL outside a trusted network: the very same exchange, inside the TLS channel. Either way a refusal is a SaslAuthenticationException — carrying the code and the message when there is one — which leaves every retry loop of the client, because nothing about the connection would be different next time. GSSAPI and the SCRAM mechanisms of 0.10.2 are refused with an explanation before a socket is opened. See examples/sasl.php, the "SASL/PLAIN" section of the protocol document and its "SaslAuthenticate API (key 36, v0)" section.

Supported Kafka protocol versions

This branch is the 4.x line, built towards the Kafka 4.3.1 wire protocol on top of the finished 3.x line (its record is docs/handoff/3.x.md). Today the client speaks Kafka 4.3 — everything Kafka 3.0 to 3.9 and 4.0 to 4.3 added, and the KIP-848 consumer protocol — against a 4.3.1 node, and the table below lists the version of every api the client sends; every minor of the line is in, and the KIP-932 share consumer is its last wave (the plan is docs/handoff/main.md). The frozen protocol snapshots of the lines below live on 3.x (Kafka 3.9.2), 2.x (Kafka 2.8.2), 1.x (Kafka 1.1.1), 0.11.x (Kafka 0.11.0.3), 0.10.x (Kafka 0.10.2.2), 0.9.x (Kafka 0.9.0.1) and 0.8.x (Kafka 0.8.2.2).

Kafka 0.10.0 added the ApiVersions request (key 18), so this line does not have to guess what its broker speaks. The "Versions served" column is the literal answer of the 4.3.1 container — a KRaft node, broker and controller in one process — on its client listeners: 75 apis, the keys 0–3, 8–51, 55, 57, 60, 61, 64–66, 68, 69, 74–81 and 83–92. The ZooKeeper apis 4–7 have no version any more, the controller-only apis 52–54, 56, 58, 59, 62, 63, 67, 70, 73 and 82 live on the controller listener, and the client-metrics apis 71 and 72 are hidden while the node has no telemetry plugin. Kafka 4.0 removed the versions below the Kafka 2.1 baseline (KIP-896): nineteen apis of the client listener lost their lowest versions and eighteen rows start above v0 (Produce's still starts at 0), and every removed version — Produce v0 to v2 too, which the answer still lists — closes the connection. The answer is read with Client::apiVersions() and pinned by tests/Integration/ApiVersionProbeTest.php, which sends one real frame of every key at its maximum version, one above it and one below every minimum.

The "main" column lists the versions this client has a class for; the one in bold is the version it sends; every version the node serves to a client is spoken, and what is left out says so in its row. The streams-group apis 88 and 89 are out by decision (the "leaves out" table below). The 2.x column is where the 3.x line started.

The lower versions of every api are kept because their frames are what the wire vectors of the lines below replay.

The three ACL apis (29, 30, 31) are implemented on this line, at the version 3 of Kafka 3.3. The lines below left them out because they do nothing on a broker without an authorizer.class.name (such a broker answers all three with the error code 54, SecurityDisabled) and every wire vector of this repository is captured from a real broker. The 3.9.2 node of this line runs the StandardAuthorizer of KRaft with super.users=User:ANONYMOUS;User:admin;User:kafkatest, so AdminClient::describeAcls(), createAcls() and deleteAcls() were measured against a real authorizer: an acl is a Common\AclBinding — a ResourcePattern (LITERAL or PREFIXED, with the USER resource type that Kafka 3.3 added for KIP-373) and an AccessControlEntry (a principal, a host, an AclOperation and ALLOW/DENY) — and a describe or a delete names an AclBindingFilter, in which every field may be a wildcard and the pattern type MATCH asks which acls apply to a resource. The client sends the version 3 and keeps no lower one: an api that starts on this line gets the versions the node was measured at.

The four delegation-token apis (38 to 41) are implemented, and a token cannot be used to authenticate. AdminClient::createDelegationToken(), renewDelegationToken(), expireDelegationToken() and describeDelegationToken() speak them over an authenticated channel — one of the SASL listeners, because KIP-48 derives the owner of a token from the principal of the connection and answers the error code 64 on a PLAINTEXT or one-way-SSL one. What is missing is the other half of KIP-48: using a token means a SASL/SCRAM login whose user name is the token id and whose password is the base64 HMAC, and this client speaks SASL/PLAIN only. The four apis are verified against the real broker, the login with their result is not implemented. examples/delegation-tokens.php runs one token's whole life against the SASL listener of docker-compose.yml.

What the lines can do beyond the api versions themselves (the main column is the state of the current milestone):

What this line leaves out by design (the owner's decisions for the 3.x and the 4.x line; everything else the 4.3.1 node serves to a client is implemented):

Five properties of a 3.9.2 node (and of every broker since 1.0) regularly surprise clients, and this implementation deals with all of them explicitly:

• An api the broker does not serve costs the connection. A request whose api key or version a 3.9.2 node cannot parse — and a body that does not match the schema of a version it does serve — makes it close the socket: Closing socket for … because of error in the broker log, and the end of the stream for the client, reported as a NetworkException. A 0.9.0.1 broker only dropped such a frame and kept the connection open, so code ported from that line waits for a timeout that will never come. There is exactly one exception: ApiVersions answers an unknown version with the error code 35 and survives. Up to Kafka 0.11 there was a second, ControlledShutdown, which the broker still parsed with a Scala class that never looked at the version; Kafka 1.0 moved that api to the schemas of the Java client, so it now serves v0 and v1 and hangs up on anything above. This client only ever sends the api versions of the table above.
• A group whose last member leaves does not disappear. Since Kafka 0.10.1 it stays in the state Empty with its committed offsets until offsets.retention.minutes expires them, is still listed by AdminClient::listGroups() and is described as Empty, not Dead. That is what lets a restarted consumer of the same group resume where the group committed — and it is a behaviour change against 0.9, where the coordinator dropped such a group at once.
• A fetch answer may be empty although the partition has data — and larger than the limit it asked for. Fetch v3 (Kafka 0.10.1) added a request-level fetch.max.bytes, and the broker spends it on the partitions in the order of the request: a partition behind an exhausted budget comes back empty with its high water mark above the fetch offset, and the next poll, which rotates the served partitions to the back, picks it up. The other half of the same rule is that the first non-empty partition is always served whole, even when the single message exceeds the limit — which is why a consumer of this branch can no longer get stuck on a record that is bigger than max.partition.fetch.bytes.
• The coordinator of a group is not available right away. The first GroupCoordinator request for any group makes the broker create the internal __consumer_offsets topic and is answered with the error code 15 while that happens; code 14 means the coordinator is still reading the offsets of the group out of it. Both are retried with retry.backoff.ms until metadata.fetch.timeout.ms by Common\CoordinatorLookup, which AdminClient::findCoordinator() and Client::getGroupCoordinator() use.
• The broker remembers the last five batches of a producer, and it drops record headers when it converts a batch down. Both are Kafka 1.0 changes against the 0.11 line and both are invisible in the frame: a duplicate of any of the last five batches of a producer id and partition is answered as the original append (0.11 remembered one batch and answered 45 for anything older), and a Fetch below v4 of a partition whose records carry headers now succeeds with the headers silently removed, where a 0.11 broker refused the whole partition with the error code -1.

One thing that a 0.10 broker no longer does: a broker without a single topic answers Metadata with its brokers, where 0.8 and 0.9 answered an empty broker array until some topic existed. An empty broker array is still "not ready, retry" and never "the cluster has no brokers", and tests/Fixture/ClusterReadinessProbe.php still treats it that way, but it is no longer the normal state of a fresh cluster.

Two changes against the 0.8.2.2 line show up in single apis, and both are worth knowing when porting code between the branches. OffsetFetch v1 no longer validates the partition: asking for a partition the cluster does not host answers offset -1 with the error code 0 — "nothing committed" — where 0.8.2.2 answered 3 (UnknownTopicOrPartition), so Metadata is the only api that says whether a partition exists. And a ControlledShutdown for a broker id the controller does not know was answered with 8 (BrokerNotAvailable) from 0.9 on, the code a 0.8.2.2 broker turned into -1 (Unknown) by mapping the cause of an exception that has none — a measurement of the lines below, since a KRaft node does not serve the api on a client listener. The protocol document has the details.

Testing & Contributing

composer install
composer check   # coding standards + static analysis + PHPUnit

The suite is split in three — 1717 unit tests, 321 compliance tests replaying the 314 documented wire vectors, and 540 integration tests against a real broker over its four listeners, without a single skip:

vendor/bin/phpunit --testsuite unit          # pure unit tests, no broker
vendor/bin/phpunit --testsuite compliance    # replays the documented wire vectors

docker compose up -d                         # Kafka 1.1.1: PLAINTEXT 9092, SSL 9093,
                                             #                 SASL_PLAINTEXT 9094, SASL_SSL 9095
KAFKA_BOOTSTRAP_SERVERS=127.0.0.1:9092 vendor/bin/phpunit --testsuite integration

The integration suite is skipped unless KAFKA_BOOTSTRAP_SERVERS points at a running broker, and each of the other three listeners has an environment variable of its own — a test that needs one is skipped when it is unset:

The compliance suite replays every wire vector of docs/protocol/vectors — frames that a real Kafka broker sent or accepted — through the request and response classes and checks that the annotated dumps of docs/protocol/4.3.md still hold the same bytes, and that every @see docs/protocol/4.3.md, section "…" of the sources names a heading that exists, so the document and the code cannot drift apart.

Examples

Every file in examples/ is runnable against the container of docker-compose.yml:

Issues and pull requests are welcome.

License

Released under the MIT license.