+
+
+
+
This extension provides ways to interact with Apache Kafka.
It also supports any system that is Kafka API compatible like Redpanda.
You can find some examples for producer and consumer here
This extension uses librdkafka for binding and was inspired by php-rdkafka.
Notice: php-rdkafka conflicts with this extension, please uninstall it first if you want to use simple-kafka-client.
+Documentation of php-simple-kafka-clientpublic function dump(): array {}
+Dumps the current configuration
$conf = new SimpleKafkaClient\Configuration();
+$conf->set('auto.offset.reset', 'earliest');
+$conf->dump();
+public function set(string $name, string $value): void {}
+Set a configuration value
$conf = new SimpleKafkaClient\Configuration();
+$conf->set('auto.offset.reset', 'earliest');
+public function setDrMsgCb(callable $callback): void {}
+Set a message delivery report callback
This callback is called exactly once per message, indicating if
the message was succesfully delivered or permanently failed delivery
$conf = new SimpleKafkaClient\Configuration();
+$conf->setDrMsgCb(
+ function (Producer $kafka, Message $message) {
+ if (RD_KAFKA_RESP_ERR_NO_ERROR !== $message->err) {
+ $errorStr = rd_kafka_err2str($message->err);
+
+ echo sprintf('Message FAILED (%s, %s) to send with payload => %s', $message->err, $errorStr, $message->payload) . PHP_EOL;
+ } else {
+ // message successfully delivered
+ echo sprintf('Message sent SUCCESSFULLY with payload => %s', $message->payload) . PHP_EOL;
+ }
+ }
+);
+public function setErrorCb(callable $callback): void {}
+Gets triggered for every error that occurs in librdkafka, keep in mind
that non-critical errors will be retried by libdrkafka
$conf = new SimpleKafkaClient\Configuration();
+$conf->setErrorCb(
+ function (SimpleKafkaClient\Kafka $kafka, $errorCode, $reason) {
+ //do something
+ }
+);
+public function setLogCb(callable $callback): void {}
+Set a log callback
You will get events according to the log_level setting
$conf = new SimpleKafkaClient\Configuration();
+$conf->setLogCb(
+ function (SimpleKafkaClient\Kafka $kafka, int $level, string $facility, string $message) {
+ //do something
+ }
+);
+public function setOAuthBearerTokenRefreshCb(callable $callback): void {}
+The SASL/OAUTHBEARER token refresh callback is triggered automatically or via poll
whenever OAUTHBEARER is the SASL mechanism and a token needs to be retrieved,
typically based on the configuration defined in sasl.oauthbearer.config.
The callback should invoke setOauthBearerToken
or setOauthBearerTokenFailure to indicate success
or failure, respectively.
$conf = new SimpleKafkaClient\Configuration();
+$conf->set('metadata.broker.list', getenv('TEST_KAFKA_BROKERS'));
+$conf->set('security.protocol', 'SASL_PLAINTEXT');
+$conf->set('sasl.mechanisms', 'OAUTHBEARER');
+$conf->set('sasl.oauthbearer.config', 'principalClaimName=azp');
+$conf->setOAuthBearerTokenRefreshCb(function($kafka, $oAuthBearerConfig) {
+ // get the refresh token with some custom code, then act accordingly
+ if ($tokenRefreshWasSucessful) {
+ $kafka->setOAuthBearerToken($token, $lifetimeMs, $principalName, $extensions);
+ } else {
+ $kafka->setOAuthBearerTokenFailure($errorReason);
+ }
+});
+public function setOffsetCommitCb(callable $callback): void {}
+Set offset commit callback for use with consumer groups.
The results of automatic or manual offset commits will be scheduled
for this callback. If no partitions had valid offsets to commit
this callback will be called with err == RD_KAFKA_RESP_ERR__NO_OFFSET
which is not to be considered an error.
$conf = new SimpleKafkaClient\Configuration();
+$conf->setOffsetCommitCb(
+ function (SimpleKafkaClient\Kafka $kafka, int $errorCode, array $topicPartition) {
+ if (RD_KAFKA_RESP_ERR_NO_ERROR === $errorCode) {
+ echo 'Commit was successful';
+ } else {
+ echo 'Commit failed';
+ }
+ }
+);
+public function setRebalanceCb(callable $callback): void {}
+Set a rebalance callback for use with coordinated consumer group balancing.
The $errorCode is set to either RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS
or RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS and ‘partitions’ contains
the full partition set that was either assigned or revoked.
Registering a rebalance callback turns off the automatic partition
assignment/revocation and instead delegates that responsibility
to the application’s callback.
$conf = new SimpleKafkaClient\Configuration();
+$conf->setRebalanceCb(
+ function (SimpleKafkaClient\Consumer $kafka, int $errorCode, array $partitions = null) {
+ case RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
+ $kafka->assign($partitions);
+ break;
+
+ case RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS:
+ if ($manual_commits) {
+ // Optional explicit manual commit
+ $kafka->commit($partitions);
+ }
+ //revoke partitions
+ $kafka->assign(NULL);
+ break;
+
+ default:
+ // handle arbitrary rebalancing failure
+ ...
+ // synchronize state
+ $kafka->assign(NULL);
+ break;
+ }
+);
+public function setStatsCb(callable $callback): void {}
+Set a statistics callback
The statistics callback is triggered every statistics.interval.ms (needs to be configured separately).
$conf = new SimpleKafkaClient\Configuration();
+$conf->setStatsCb(
+ function (SimpleKafkaClient\Kafka $kafka, string $json, int $jsonLength) {
+ //do something
+ }
+);
+public function __construct(Configuration $configuration) {}
+Get a consumer instance
$conf = SimpleKafkaClient\Configuration();
+$conf->set('metadata.broker.list', 'kafka:9092');
+$consumer = new SimpleKafkaClient\Consumer($conf);
+public function assign(array $topics): void {}
+Atomic assignment of partitions to consume.
The new partitions will replace the existing assignment.
$conf = SimpleKafkaClient\Configuration();
+$conf->set('metadata.broker.list', 'kafka:9092');
+$consumer = new SimpleKafkaClient\Consumer($conf);
+$consumer->assign(
+ [
+ new SimpleKafkaClient\TopicPartition('test-topic', 1, 3000),
+ new SimpleKafkaClient\TopicPartition('test-topic', 2, 3009)
+ ]
+);
+public function close(): void {}
+Close down the Consumer. This call will block until
the consumer has revoked its assignment, calling the rebalance callback
if it is configured, committed offsets to broker, and left the consumer group. The maximum blocking time is roughly limited to session.timeout.ms.
$conf = SimpleKafkaClient\Configuration();
+$conf->set('metadata.broker.list', 'kafka:9092');
+$consumer = new SimpleKafkaClient\Consumer($conf);
+$consumer->close();
+/**
+ * @throws SimpleKafkaClient\Exception
+ */
+public function commit($messageOrOffsets): void {}
+Commit offsets synchronously, block until offsets are
committed or the commit fails and an exception is thrown.
null is passed, latest offsets for the current assignment will be committedSimpleKafkaClient\Message is passed, commit offset for a single topic+partition based on the messageSimpleKafkaClient\TopicPartition is passed, commit offsets for the provided list of partitions$conf = SimpleKafkaClient\Configuration();
+$conf->set('metadata.broker.list', 'kafka:9092');
+$consumer = new SimpleKafkaClient\Consumer($conf);
+$message = $consumer->consume(20000);
+
+if (RD_KAFKA_RESP_ERR_NO_ERROR !== $message->err) {
+ echo 'An error occured:' . rd_kafka_err2str($message->err) . PHP_EOL;
+ return;
+}
+
+$consumer->commit($message);
+public function commitAsync($messageOrOffsets): void {}
+Commit offsets asynchronously
null is passed, latest offsets for the current assignment will be committedSimpleKafkaClient\Message is passed, commit offset for a single topic+partition based on the messageSimpleKafkaClient\TopicPartition is passed, commit offsets for the provided list of partitions$conf = SimpleKafkaClient\Configuration();
+$conf->set('metadata.broker.list', 'kafka:9092');
+$consumer = new SimpleKafkaClient\Consumer($conf);
+$message = $consumer->consume(20000);
+
+if (RD_KAFKA_RESP_ERR_NO_ERROR !== $message->err) {
+ echo 'An error occured:' . rd_kafka_err2str($message->err) . PHP_EOL;
+ return;
+}
+
+$consumer->commitAsync($message);
+public function consume(int $timeoutMs): Message {}
+Consume message(s) (will also get error events and triggers callbacks)
Registered callbacks will be automaically called rebalanceCallback, logCallback, etc.
On error $message->err will not be RD_KAFKA_ERR_NO_ERROR but contain the acutal error code.
$conf = SimpleKafkaClient\Configuration();
+$conf->set('metadata.broker.list', 'kafka:9092');
+$consumer = new SimpleKafkaClient\Consumer($conf);
+$message = $consumer->consume(20000);
+An application should call consume() at regular intervals, even if no messages
are expected, to serve any queued callbacks waiting to be called.
public function getAssignment(): array {}
+Returns the current partition assignment
$conf = SimpleKafkaClient\Configuration();
+$conf->set('metadata.broker.list', 'kafka:9092');
+$consumer = new SimpleKafkaClient\Consumer($conf);
+$consumer->assign(
+ [
+ new SimpleKafkaClient\TopicPartition('test-topic', 1, 3000),
+ new SimpleKafkaClient\TopicPartition('test-topic', 2, 3009)
+ ]
+);
+var_dump($consumer->getAssignment());
+public function getCommittedOffsets(array $topics, int $timeoutMs): array {}
+Returns the committed offsets for topics and partitions for a consumer group
$conf = SimpleKafkaClient\Configuration();
+$conf->set('metadata.broker.list', 'kafka:9092');
+$consumer = new SimpleKafkaClient\Consumer($conf);
+$topicPartition = new TopicPartition('test-topic', 0);
+var_dump($consumer->getCommittedOffsets([$topicPartition], 10000));
+function getMetadata(bool $allTopics, int $timeoutMs, ConsumerTopic $topic = null): Metadata {}
+Get metadata for all topics or a single topic
$conf = SimpleKafkaClient\Configuration();
+$conf->set('metadata.broker.list', 'kafka:9092');
+$consumer = new SimpleKafkaClient\Consumer($conf);
+$topicHandle = $consumer->getTopicHandle('test-topic');
+$singleTopicMetadata = $consumer->metadata(true, $topicHandle, 10000);
+public function getOffsetPositions(array $topics): array {}
+The offset field of each requested partition will be set to the offset of the last consumed message + 1
If there was no previous message RD_KAFKA_OFFSET_INVALID will be returned
$conf = SimpleKafkaClient\Configuration();
+$conf->set('metadata.broker.list', 'kafka:9092');
+$consumer = new SimpleKafkaClient\Consumer($conf);
+$topicPartition = new TopicPartition('test-topic', 0);
+$topicPartitionsWithOffsets = $consumer->getOffsetPositions([$topicPartition]));
+public function getSubscription(): array {}
+Return topic names to which the consumer is currently subscribed to
$conf = SimpleKafkaClient\Configuration();
+$conf->set('metadata.broker.list', 'kafka:9092');
+$consumer = new SimpleKafkaClient\Consumer($conf);
+$consumer->subscribe(['test-topic']);
+var_dump($consumer->getSubscription());
+public function getTopicHandle(string $topic): ConsumerTopic {}
+Get a topic handle for a given topic name. A topic handle is needed
for example to query metadata from the broker
$conf = SimpleKafkaClient\Configuration();
+$conf->set('metadata.broker.list', 'kafka:9092');
+$consumer = new SimpleKafkaClient\Consumer($conf);
+$topicHandle = $consumer->getTopicHandle('test-topic');
+
+// use the topic handle for further calls, e.g. to query metadata
+$singleTopicMetadata = $consumer->metadata(true, $topicHandle, 10000);
+public function offsetsForTimes(array $topicPartitions, int $timeoutMs): array {}
+Look up the offsets for the given partitions by timestamp.
The returned offset for each partition is the earliest offset whose
timestamp is greater than or equal to the given timestamp in the
corresponding partition.
$conf = SimpleKafkaClient\Configuration();
+$conf->set('metadata.broker.list', 'kafka:9092');
+$consumer = new SimpleKafkaClient\Consumer($conf);
+$topicPartition = new TopicPartition('test-topic', 0, strtotime("-1 week"));
+$offsetsOneWeekAgo = $consumer->offsetForTimes([$topicPartition], 10000);
+public function queryWatermarkOffsets(string $topic, int $partition, int &$low, int &$high, int $timeoutMs): void {}
+Query broker for low (oldest) and high (newest) offsets for a partition
$low = 0;
+$high = 0;
+
+$conf = SimpleKafkaClient\Configuration();
+$conf->set('metadata.broker.list', 'kafka:9092');
+$consumer = new SimpleKafkaClient\Consumer($conf);
+$topicPartition = new TopicPartition('test-topic', 0, strtotime("-1 week"));
+$consumer->queryWatermarkOffsets('test-topic', 0, int &$low, int &$high, 10000);
+public function setOAuthBearerToken(string $token, int $lifetimeMs, string $principalName, ?array $extensions = null): void {}
+The SASL/OAUTHBEARER token refresh callback or event handler should invoke
this method upon success. The extension keys must not include the reserved
key auth, and all extension keys and values must conform to the required
format as per https://tools.ietf.org/html/rfc7628#section-3.1
$conf = new SimpleKafkaClient\Configuration();
+$conf->set('metadata.broker.list', getenv('TEST_KAFKA_BROKERS'));
+$conf->set('security.protocol', 'SASL_PLAINTEXT');
+$conf->set('sasl.mechanisms', 'OAUTHBEARER');
+$conf->set('sasl.oauthbearer.config', 'principalClaimName=azp');
+$conf->setOAuthBearerTokenRefreshCb(function($kafka, $oAuthBearerConfig) {
+ // get the refresh token with some custom code, then act accordingly
+ if ($tokenRefreshWasSucessful) {
+ $kafka->setOAuthBearerToken($token, $lifetimeMs, $principalName, $extensions);
+ } else {
+ $kafka->setOAuthBearerTokenFailure($errorReason);
+ }
+});
+public function setOauthBearerTokenFailure(string $errorString): void {}
+The SASL/OAUTHBEARER token refresh callback or event handler should invoke
this method upon failure. $errorString should be a human readable error reason
why acquiring a token failed.
$conf = new SimpleKafkaClient\Configuration();
+$conf->set('metadata.broker.list', getenv('TEST_KAFKA_BROKERS'));
+$conf->set('security.protocol', 'SASL_PLAINTEXT');
+$conf->set('sasl.mechanisms', 'OAUTHBEARER');
+$conf->set('sasl.oauthbearer.config', 'principalClaimName=azp');
+$conf->setOAuthBearerTokenRefreshCb(function($kafka, $oAuthBearerConfig) {
+ // get the refresh token with some custom code, then act accordingly
+ if ($tokenRefreshWasSucessful) {
+ $kafka->setOAuthBearerToken($token, $lifetimeMs, $principalName, $extensions);
+ } else {
+ $kafka->setOAuthBearerTokenFailure($errorReason);
+ }
+});
+public function subscribe(array $topics): void {}
+Subscribe to one or more topics (regexp also supported).
Any previous subscription will be unassigned and unsubscribed first.
$conf = SimpleKafkaClient\Configuration();
+$conf->set('metadata.broker.list', 'kafka:9092');
+$consumer = new SimpleKafkaClient\Consumer($conf);
+$consumer->subscribe(['test-topic']);
+public function unsubscribe(): void {}
+Unsubscribe from the current subscriptions
$conf = SimpleKafkaClient\Configuration();
+$conf->set('metadata.broker.list', 'kafka:9092');
+$consumer = new SimpleKafkaClient\Consumer($conf);
+$consumer->subscribe(['test-topic']);
+$consumer->unsubscribe();
+function kafka_err2name(int $errorCode): string {}
+Returns the name of the error
php> echo kafka_err2name(88);
+UNSTABLE_OFFSET_COMMIT
+function kafka_err2str(int $errorCode): string {}
+Returns the error message of an error code
php> echo kafka_err2str(88);
+Broker: There are unstable offsets that need to be cleared
+function kafka_get_err_descs(): array {}
+Returns a full list of error codes and their description
[
+ [
+ "code" => 88,
+ "name" => "UNSTABLE_OFFSET_COMMIT",
+ "desc" => "Broker: There are unstable offsets that need to be cleared"
+ ],
+ ...
+]
+function kafka_offset_tail(int $offset): int {}
+Returns an offset value that is $offset before the tail of the topic
function kafka_thread_cnt(): int {}
+Retrieve the current number of threads in use by librdkafka
+
+
+
+
This extension provides ways to interact with Apache Kafka.
It also supports any system that is Kafka API compatible like Redpanda.
You can find some examples for producer and consumer here
This extension uses librdkafka for binding and was inspired by php-rdkafka.
Notice: php-rdkafka conflicts with this extension, please uninstall it first if you want to use simple-kafka-client.
pecl install simple_kafka_client
+For PHP7:
apk add php7-pecl-simple-kafka-client --update-cache --repository http://dl-3.alpinelinux.org/alpine/edge/testing/
+For PHP8:
apk add php8-pecl-simple-kafka-client --update-cache --repository http://dl-3.alpinelinux.org/alpine/edge/testing/
+You can find the DLLs on PECL as well. If you experience an error similar to this:
The procedure entry point rd_kafka_abort_transaction could not be located in the dynamic library librdkafka.dll
+librdkaka in use. Note that librdkaka.dll needs to be in the PATH.librdkafka.dll is packaged during the PECL build.librdkafka.dll hereruntimes sub-folder.git clone https://github.com/php-kafka/php-simple-kafka-client.git
+cd php-simple-kafka-client
+phpize && ./configure && make -j5 all && make install
+In your php/conf.d folder add a simple_kafka_client.ini with the following:
extension=simple_kafka_client.so
+