phpkafka

longlang/phpkafka is a PHP Kafka client designed to work in both PHP-FPM and Swoole environments. It is built based on the JSON protocol structures from the Java version, allowing it to support a wide range of Kafka APIs.

Key Features:

• API Support: Supports all 50 Kafka APIs.
• Compression: Supports gzip, snappy, lz4, and zstd message compression.
• Environment Compatibility: Intelligent detection and compatibility for both PHP-FPM and Swoole.
• Security: Supports SASL authentication and SSL encrypted communication.
• Core Components: Includes Producer and Consumer classes.

The Kafka protocol is defined via JSON specification files that describe how clients and servers exchange information. The protocol relies on Requests and Responses, where each request is identified by a 16-bit api key.

Key concepts include:

• Versions: Each message has a unique 16-bit version number. The version of a response must always match the version of the corresponding request.
• validVersions: A top-level field in requests/responses specifying the range of protocol versions supported (e.g., "0-2" means versions 0, 1, and 2 are understood).
• MessageData Objects: These are generated objects that represent all possible versions of a message, allowing a single code path to handle multiple versions.
• Flexible Versions: Newer versions that use more efficient serialization (e.g., COMPACT_STRING instead of STRING) to save space.

The producer uses a default partitioning strategy if no specific partition is provided in the message. The logic follows this hierarchy:

1. Explicit Partition: If a partition is specified, that partition is used.
2. Key-based Partitioning: If partition is null but a key is provided, the partition is selected using crc32(key) % partitions.
3. Round Robin: If both partition and key are null, the producer uses a Round Robin strategy to select a partition.

To install the longlang/phpkafka client, use Composer to add it to your project dependencies.

composer require longlang/phpkafka

Before installing, ensure your environment meets the following requirements:

• PHP: >= 7.1
• Kafka: >= 1.0.0
• Swoole: >= 4.5 (optional, required if using Swoole-specific features)

To maintain compatibility between clients and servers, avoid the following changes to the message protocol:

1. Modifying Released Versions: Never change a protocol version that has already been released. Correct mistakes by issuing a new version.
2. Re-ordering Fields: You may add fields, but you must never change the relative order of existing fields.
3. Changing Defaults: Never change the default value of an existing field.
4. Changing Types: Never change the type of an existing field (except for converting an array of primitives to an array of single-field structures, which is binary-compatible).

To use SASL authentication, pass an array to $config->setSasl(). The type field should specify the SASL Authentication Type, such as \longlang\phpkafka\Sasl\PlainSasl::class for PLAIN authentication.

use longlang//
use longlang//

$config = new ConsumerConfig();
// .... Your Other Config
$config->setSasl([
    "type"=>\longlang\phpkafka\Sasl\PlainSasl::class,
    "username"=>"admin",
    "password"=>"admin-secret"
]);
$consumer = new Consumer($config);
// .... Your Business Code

To authenticate with Kafka using SASL, pass a configuration array to setSasl() in ProducerConfig.

Supported SASL types include \longlang\phpkafka\Sasl\PlainSasl::class for PLAIN authentication. You must provide the type, username, and password.

use longlang\phpkafka\Producer\ProducerConfig;
use longlang\phpkafka\Producer\Producer;

$config = new ProducerConfig();
$config->setSasl([
    "type" => \longlang\phpkafka\Sasl\PlainSasl::class,
    "username" => "admin",
    "password" => "admin-secret"
]);
$producer = new Producer($config);

Use longlang\phpkafka\Config\SslConfig to configure SSL/TLS settings and pass it to the ProducerConfig via setSsl().

use longlang\phpkafka\Producer\ProducerConfig;
use longlang\phpkafka\Producer\Producer;
use longlang\phpkafka\Config\SslConfig;

$config = new ProducerConfig();
$sslConfig = new SslConfig();
$sslConfig->setOpen(true);
$sslConfig->setVerifyPeer(true);
$sslConfig->setAllowSelfSigned(true);
$sslConfig->setCafile("/kafka-client/.github/kafka/cert/ca-cert");

$config->setSsl($sslConfig);
$producer = new Producer($config);