Winter Kafka: Kafka Producer and Consumer Module

Winter Kafka connects Winter Boot applications to Apache Kafka. It wires producers and consumer workers into the application lifecycle, so you send and receive events with YAML configuration plus a small worker class.

Install the prerequisites

Winter Kafka needs the swoole and rdkafka PHP extensions plus the modules package. Follow the Installation guide (Docker is recommended) to set them up.
Enable in application.yml

Register the module and point it to your config file.
application.yml
```
modules:
  - module: dev\winterframework\kafka\KafkaModule
    enabled: true
    configFile: kafka-config.yml
```
configFile is a file path relative to your config directory or an absolute path.

kafka-config.yml

The Kafka config declares your bootstrap servers, consumer groups, and named producers. You can apply global defaults with the reserved __default__ name and override per group or producer as needed.

bootstrap.servers: 127.0.0.1:9092

# Consumer configuration and groups
consumers:
  - name: __default__
    transientExceptions: []

  - name: group1-name
    topics: [topic1-name, topic99-name]
    workerNum: 1
    workerClass: some\package\className

  - name: group2-name
    topics: [topic2-name]
    workerNum: 1
    workerClass: some\package2\className2

# Producer configuration
producers:
  - name: __default__
    message.max.bytes: 10485760
    retries: 5

  - name: p-group1
    topic: p-topic1

  - name: p-group2
    topic: p-topic2

Autowired services

Once the module is enabled, KafkaService can be autowired. Consumer workers start automatically when the application boots, and every configured producer is ready to use.

The reserved "__default__" consumer and producer entries can contain any rdkafka settings. Defaults apply to all producers and consumers; each individual entry may also specify its own rdkafka settings, which apply only to that producer or consumer.

use dev\winterframework\kafka\KafkaService;
use dev\winterframework\stereotype\Autowired;

#[Autowired]
private KafkaService $kafkaService;

// produce(producerName, message, key): sends to the producer's configured topic
$this->kafkaService->produce('p-group1', json_encode(['orderId' => 42]), 'order-42');

Consumer properties

Name	Required?	Description
name	Yes	Consumer group name
topic	Yes	(string) (alias to `topics`) Topic to consume. Either `topic` or `topics` is needed.
topics	Yes	(array) (alias to `topic`) Array of topics to consume. Either `topic` or `topics` is needed.
workerNum	No	Default value: `1`. Number of consumer worker processes to start.
workerClass	Yes	Actual consumer class implementation. Must be derived from `AbstractConsumer`.
statsCallback	No	Default value: `dev\winterframework\kafka\consumer\ConsumerLagMonitor`. Sets the statistics report callback; must be derived from `ConsumerStatisticsCallback`. Set `0` to disable.
errorCallback	No	Default value: `dev\winterframework\kafka\consumer\ConsumerErrorCallbackDefault`. Sets the error callback, must be derived from `ConsumerErrorCallback`. Set `0` to disable.
log_level	No	Default value: `6` (integer). Set log level value. EMERG = 0, ALERT = 1, CRIT = 2, ERR = 3, WARNING = 4, NOTICE = 5, INFO = 6, DEBUG = 7.
logCallback	No	Default value: `dev\winterframework\kafka\KafkaLogCallbackDefault`. Sets the log callback; you will receive events according to `log_level`. Must be derived from `KafkaLogCallback`. Set `0` to disable.
offsetCommitCallback	No	Default value: `(empty)`. Sets the offset commit callback for use with consumer groups. Must be derived from `ConsumerOffsetCommitCallback`.
rebalanceCallback	No	Default value: `dev\winterframework\kafka\consumer\ConsumerRebalanceCallbackDefault`. Sets the rebalance callback for coordinated consumer group balancing. Must be derived from `ConsumerRebalanceCallback`.
(Others)	-	Full list of settings

Producer properties

Name	Required	Description
name	Yes	Producer name
topic	Yes	(string) Message will be produced to this topic.
log_level	No	Default value: `6` (integer). Set log level value. EMERG = 0, ALERT = 1, CRIT = 2, ERR = 3, WARNING = 4, NOTICE = 5, INFO = 6, DEBUG = 7.
logCallback	No	Default value: `dev\winterframework\kafka\KafkaLogCallbackDefault`. Sets the log callback; you will receive events according to `log_level`. Must be derived from `KafkaLogCallback`. Set `0` to disable.
(Others)	-	Full list of settings

How to write a consumer worker

A consumer worker must implement the Consumer interface. In practice, extend AbstractConsumer and implement the consume method.

use dev\winterframework\kafka\consumer\AbstractConsumer;
use dev\winterframework\kafka\consumer\ConsumerRecord;
use dev\winterframework\kafka\consumer\ConsumerRecords;

class TestKafkaConsumer extends AbstractConsumer {

    public function consume(ConsumerRecords $records): void {

        foreach ($records as $record) {

            // Do something with ConsumerRecord

            /** @var ConsumerRecord $record */
            self::logInfo("Received a message "
                . ' Message: ' . $record->getValue()
                . ', Topic: ' . $record->getTopic()
                . ', Group: ' . $record->getGroupName()
                . ', Time: ' . $record->getTimestamp()
                . ', Headers: ' . json_encode($record->getHeaders())
                . ', Offset: ' . $record->getOffset()
                . ', Partition: ' . $record->getPartition()
            );

        }

    }

}