### Install Laravel Kafka via Composer

Source: https://laravelkafka.com/docs/v2.0/installation-and-setup

Installs the Laravel Kafka package into your project using Composer. This command adds the necessary dependencies to your project's composer.json file.

```shell
composer require mateusjunges/laravel-kafka
```

--------------------------------

### Install Supervisor

Source: https://laravelkafka.com/docs/v2.0/consuming-messages/class-structure

Commands to install the Supervisor process monitor on Ubuntu and macOS.

```bash
sudo apt-get install supervisor
```

```bash
brew install supervisor
```

--------------------------------

### Publish Laravel Kafka Configuration

Source: https://laravelkafka.com/docs/v2.0/installation-and-setup

Publishes the default configuration file for Laravel Kafka to your Laravel project. This allows you to customize Kafka settings like brokers, consumer groups, and more.

```shell
php artisan vendor:publish --tag=laravel-kafka-config
```

--------------------------------

### Default Laravel Kafka Configuration File

Source: https://laravelkafka.com/docs/v2.0/installation-and-setup

The default configuration array for Laravel Kafka, defining essential settings such as Kafka brokers, consumer group IDs, offset reset policies, auto-commit behavior, compression codecs, and debug modes.

```php
<?php declare(strict_types=1);

return [
    /*
     | Your kafka brokers url.
     */
    'brokers' => env('KAFKA_BROKERS', 'localhost:9092'),

    /*
     | Kafka consumers belonging to the same consumer group share a group id.
     | The consumers in a group then divides the topic partitions as fairly amongst themselves as possible by
     | establishing that each partition is only consumed by a single consumer from the group.
     | This config defines the consumer group id you want to use for your project.
     */
    'consumer_group_id' => env('KAFKA_CONSUMER_GROUP_ID', 'group'),

    'consumer_timeout_ms' => env("KAFKA_CONSUMER_DEFAULT_TIMEOUT", 2000),

    /*
     | After the consumer receives its assignment from the coordinator,
     | it must determine the initial position for each assigned partition.
     | When the group is first created, before any messages have been consumed, the position is set according to a configurable
     | offset reset policy (auto.offset.reset). Typically, consumption starts either at the earliest offset or the latest offset.
     | You can choose between "latest", "earliest" or "none".
     */
    'offset_reset' => env('KAFKA_OFFSET_RESET', 'latest'),

    /*
     | If you set enable.auto.commit (which is the default), then the consumer will automatically commit offsets periodically at the
     | interval set by auto.commit.interval.ms.
     */
    'auto_commit' => env('KAFKA_AUTO_COMMIT', true),

    'sleep_on_error' => env('KAFKA_ERROR_SLEEP', 5),

    'partition' => env('KAFKA_PARTITION', 0),

    /*
     | Kafka supports 4 compression codecs: none , gzip , lz4 and snappy
     */
    'compression' => env('KAFKA_COMPRESSION_TYPE', 'snappy'),

    /*
     | Choose if debug is enabled or not.
     */
    'debug' => env('KAFKA_DEBUG', false),

    /*
     | Repository for batching messages together
     | Implement BatchRepositoryInterface to save batches in different storage
     */
    'batch_repository' => env('KAFKA_BATCH_REPOSITORY', \Junges\Kafka\BatchRepositories\InMemoryBatchRepository::class),

    /*
     | The sleep time in milliseconds that will be used when retrying flush
     */
    'flush_retry_sleep_in_ms' => 100,

    /*
     | The cache driver that will be used
     */
    'cache_driver' => env('KAFKA_CACHE_DRIVER', env('CACHE_DRIVER', 'file')),
];

```

--------------------------------

### ksqldb-cli Docker Service

Source: https://laravelkafka.com/docs/v2.0/example-docker-compose

Defines the ksqlDB CLI service for a Docker Compose setup, specifying the image, container name, and dependencies.

```yaml
ksqldb-cli:
  image: confluentinc/cp-ksqldb-cli:7.3.2
  container_name: ksqldb-cli
  depends_on:
    - broker
    - connect
    - ksqldb-server
  entrypoint: /bin/sh
  tty: true
```

--------------------------------

### Laravel Kafka Consumer Example

Source: https://laravelkafka.com/docs/v2.0/consuming-messages/class-structure

Example of a Laravel command class to consume messages from a Kafka topic using the JungesKafka package. It sets up brokers, auto-commit, and a message handler.

```php
<?php declare(strict_types=1);

namespace App\Console\Commands\Consumers;

use Illuminate\Console\Command;
use Junges\Kafka\Facades\Kafka;
use Junges\Kafka\Contracts\MessageConsumer;
use Junges\Kafka\Contracts\ConsumerMessage;

class MyTopicConsumer extends Command
{
    protected $signature = "consume:my-topic";

    protected $description = "Consume Kafka messages from 'my-topic'.";

    public function handle()
    {
        $consumer = Kafka::consumer(['my-topic'])
            ->withBrokers('localhost:8092')
            ->withAutoCommit()
            ->withHandler(function(ConsumerMessage $message, MessageConsumer $consumer) {
                // Handle your message here
            })
            ->build();

            $consumer->consume();
    }
}
```

--------------------------------

### Docker Compose Setup for Kafka Cluster

Source: https://laravelkafka.com/docs/v2.0/example-docker-compose

This `docker-compose.yml` file defines the services required to run a local Kafka development environment. It orchestrates Zookeeper, Kafka Broker, Schema Registry, Kafka Connect, Control Center, and ksqlDB, specifying their Docker images, ports, environment variables, and inter-service dependencies.

```yaml
version: '2'
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.3.2
    hostname: zookeeper
    container_name: zookeeper
    ports:
      - "2181:2181"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  broker:
    image: confluentinc/cp-server:7.3.2
    hostname: broker
    container_name: broker
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
      - "9101:9101"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://broker:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_METRIC_REPORTERS: io.confluent.metrics.reporter.ConfluentMetricsReporter
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      KAFKA_CONFLUENT_LICENSE_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_CONFLUENT_BALANCER_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_JMX_PORT: 9101
      KAFKA_JMX_HOSTNAME: localhost
      KAFKA_CONFLUENT_SCHEMA_REGISTRY_URL: http://schema-registry:8081
      CONFLUENT_METRICS_REPORTER_BOOTSTRAP_SERVERS: broker:29092
      CONFLUENT_METRICS_REPORTER_TOPIC_REPLICAS: 1
      CONFLUENT_METRICS_ENABLE: 'true'
      CONFLUENT_SUPPORT_CUSTOMER_ID: 'anonymous'

  schema-registry:
    image: confluentinc/cp-schema-registry:7.3.2
    hostname: schema-registry
    container_name: schema-registry
    depends_on:
      - broker
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: 'broker:29092'
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081

  connect:
    image: cnfldemos/cp-server-connect-datagen:0.5.3-7.1.0
    hostname: connect
    container_name: connect
    depends_on:
      - broker
      - schema-registry
    ports:
      - "8083:8083"
    environment:
      CONNECT_BOOTSTRAP_SERVERS: 'broker:29092'
      CONNECT_REST_ADVERTISED_HOST_NAME: connect
      CONNECT_GROUP_ID: compose-connect-group
      CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs
      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_OFFSET_FLUSH_INTERVAL_MS: 10000
      CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_KEY_CONVERTER: org.apache.kafka.connect.storage.StringConverter
      CONNECT_VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter
      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081
      # CLASSPATH required due to CC-2422
      CLASSPATH: /usr/share/java/monitoring-interceptors/monitoring-interceptors-7.3.2.jar
      CONNECT_PRODUCER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor"
      CONNECT_CONSUMER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor"
      CONNECT_PLUGIN_PATH: "/usr/share/java,/usr/share/confluent-hub-components"
      CONNECT_LOG4J_LOGGERS: org.apache.zookeeper=ERROR,org.I0Itec.zkclient=ERROR,org.reflections=ERROR

  control-center:
    image: confluentinc/cp-enterprise-control-center:7.3.2
    hostname: control-center
    container_name: control-center
    depends_on:
      - broker
      - schema-registry
      - connect
      - ksqldb-server
    ports:
      - "9021:9021"
    environment:
      CONTROL_CENTER_BOOTSTRAP_SERVERS: 'broker:29092'
      CONTROL_CENTER_CONNECT_CONNECT-DEFAULT_CLUSTER: 'connect:8083'
      CONTROL_CENTER_KSQL_KSQLDB1_URL: "http://ksqldb-server:8088"
      CONTROL_CENTER_KSQL_KSQLDB1_ADVERTISED_URL: "http://localhost:8088"
      CONTROL_CENTER_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"
      CONTROL_CENTER_REPLICATION_FACTOR: 1
      CONTROL_CENTER_INTERNAL_TOPICS_PARTITIONS: 1
      CONTROL_CENTER_MONITORING_INTERCEPTOR_TOPIC_PARTITIONS: 1
      CONFLUENT_METRICS_TOPIC_REPLICATION: 1
      PORT: 9021

  ksqldb-server:
    image: confluentinc/cp-ksqldb-server:7.3.2
    hostname: ksqldb-server
    container_name: ksqldb-server
    depends_on:
      - broker
      - connect
    ports:
      - "8088:8088"
    environment:
      KSQL_CONFIG_DIR: "/etc/ksql"
      KSQL_BOOTSTRAP_SERVERS: "broker:29092"
      KSQL_HOST_NAME: ksqldb-server
      KSQL_LISTENERS: "http://0.0.0.0:8088"
      KSQL_CACHE_MAX_BYTES_BUFFERING: 0
      KSQL_KSQL_SCHEMA_REGISTRY_URL: "http://schema-registry:8081"

```

--------------------------------

### Supervisorctl Commands for Laravel Kafka Consumers

Source: https://laravelkafka.com/docs/v2.0/consuming-messages/class-structure

This section provides essential Supervisorctl commands for managing Laravel Kafka consumers. `supervisorctl reread` reloads the configuration, `supervisorctl update` applies changes, and `supervisorctl start my-topic-consumer:*` starts all processes prefixed with 'my-topic-consumer'.

```shell
sudo supervisorctl reread

sudo supervisorctl update

sudo supervisorctl start my-topic-consumer:*
```

--------------------------------

### Laravel Kafka: Before and After Consuming Callbacks

Source: https://laravelkafka.com/docs/v2.0/advanced-usage/before-callbacks

This snippet demonstrates how to configure `beforeConsuming` and `afterConsuming` callbacks for a Laravel Kafka consumer. The `beforeConsuming` callback example shows how to pause consumption if the application is in maintenance mode. These callbacks are executed sequentially and receive the `MessageConsumer` instance.

```PHP
$consumer = \Junges\Kafka\Facades\Kafka::consumer()
    ->beforeConsuming(function(\Junges\Kafka\Contracts\MessageConsumer $consumer) {
        while (app()->isDownForMaintenance()) {
            sleep(1);
        }
    })
    ->afterConsuming(function (\Junges\Kafka\Contracts\MessageConsumer $consumer) {
        // Runs after consuming the message
    });

```

--------------------------------

### Consume from Specific Kafka Offsets with Laravel Kafka

Source: https://laravelkafka.com/docs/v2.0/consuming-messages/consuming-from-specific-offsets

This snippet demonstrates how to configure a Kafka consumer to start reading messages from a specific partition and offset within a topic. It utilizes the `assignPartitions` method on the `ConsumerBuilder` and the `RdKafkaTopicPartition` class to define the starting point for consumption.

```PHP
$partition = 1; // The partition number you want to assign.
$offset = 0; // The offset you want to start consuming messages from.

$consumer = \Junges\Kafka\Facades\Kafka::consumer()
    ->assignPartitions([
        new \RdKafka\TopicPartition('your-topic-name', $partition, $offset)
    ]);

```

--------------------------------

### Configure Supervisor for Laravel Kafka Consumer

Source: https://laravelkafka.com/docs/v2.0/consuming-messages/class-structure

Example Supervisor configuration file to monitor and manage a Laravel Kafka consumer process, ensuring it restarts automatically.

```ini
[program:my-topic-consumer]
directory=/var/www/html
process_name=%(program_name)s_%(process_num)02d
command=php artisan consume:my-topic
autostart=true
autorestart=true
redirect_stderr=true
stdout_logfile=/var/log/supervisor-laravel-worker.log
stopwaitsecs=3600
```

--------------------------------

### Laravel Kafka v2.x Upgrade Guide & API Changes

Source: https://laravelkafka.com/docs/v2.0/upgrade-guide

Details significant changes in Laravel Kafka v2.x compared to v1.13.x, including renamed contracts and updated method signatures. This section helps users migrate their existing Kafka integrations.

```APIDOC
Contract Renames (v1.13.x -> v2.x):
- \Junges\Kafka\Contracts\CanProduceMessages -> \Junges\Kafka\Contracts\MessageProducer
- \Junges\Kafka\Contracts\KafkaProducerMessage -> \Junges\Kafka\Contracts\ProducerMessage
- \Junges\Kafka\Contracts\CanConsumeMessages -> \Junges\Kafka\Contracts\MessageConsumer
- \Junges\Kafka\Contracts\KafkaConsumerMessage -> \Junges\Kafka\Contracts\ConsumerMessage
- \Junges\Kafka\Contracts\CanPublishMessagesToKafka -> Removed
- \Junges\Kafka\Contracts\CanConsumeMessagesFromKafka -> Removed
- \Junges\Kafka\Contracts\CanConsumeBatchMessages -> \Junges\Kafka\Contracts\BatchMessageConsumer
- \Junges\Kafka\Contracts\CanConsumeMessages -> \Junges\Kafka\Contracts\MessageConsumer
- Introduced new: \Junges\Kafka\Contracts\Manager used by \Junges\Kafka\Factory class
```

```APIDOC
Method Signature Change: withSasl

Description: The `withSasl` method signature was changed to accept all SASL parameters directly instead of a Sasl object.

Signature:
public function withSasl(string $username, string $password, string $mechanisms, string $securityProtocol = 'SASL_PLAINTEXT');

Parameters:
- username: The SASL username.
- password: The SASL password.
- mechanisms: The SASL mechanisms (e.g., 'PLAIN', 'SCRAM-SHA-256').
- securityProtocol: The security protocol to use (defaults to 'SASL_PLAINTEXT').
```

--------------------------------

### Configure AVRO Deserializer for Kafka Consumer

Source: https://laravelkafka.com/docs/v2.0/consuming-messages/custom-deserializers

Demonstrates setting up an AVRO deserializer for Kafka consumers. It involves configuring a schema registry client, mapping AVRO schemas to Kafka topics for both keys and bodies, and initializing the Kafka consumer with the custom AVRO deserializer. This setup requires the FlixTech AvroSerializer and SchemaRegistryApi packages.

```PHP
use FlixTech\AvroSerializer\Objects\RecordSerializer;
use FlixTech\SchemaRegistryApi\Registry\CachedRegistry;
use FlixTech\SchemaRegistryApi\Registry\BlockingRegistry;
use FlixTech\SchemaRegistryApi\Registry\PromisingRegistry;
use FlixTech\SchemaRegistryApi\Registry\Cache\AvroObjectCacheAdapter;
use GuzzleHttp\Client;

// Configure the schema registry client
$cachedRegistry = new CachedRegistry(
    new BlockingRegistry(
        new PromisingRegistry(
            new Client(['base_uri' => 'kafka-schema-registry:9081'])
        )
    ),
    new AvroObjectCacheAdapter()
);

// Create the AvroSchemaRegistry and RecordSerializer instances
$registry = new \Junges\Kafka\Message\Registry\AvroSchemaRegistry($cachedRegistry);
$recordSerializer = new RecordSerializer($cachedRegistry);

// Map schemas to topics (latest version used if not specified)
$registry->addBodySchemaMappingForTopic(
    'test-topic',
    new \Junges\Kafka\Message\KafkaAvroSchema('bodySchema', 9)
);
$registry->addKeySchemaMappingForTopic(
    'test-topic',
    new \Junges\Kafka\Message\KafkaAvroSchema('keySchema', 9)
);

// Initialize the AvroDeserializer
// By default, both key and body are decoded. Use AvroDecoderInterface::DECODE_BODY or AvroDecoderInterface::DECODE_KEY to specify.
$deserializer = new \Junges\Kafka\Message\Deserializers\AvroDeserializer($registry, $recordSerializer);

// Configure the Kafka consumer to use the AVRO deserializer
$consumer = \Junges\Kafka\Facades\Kafka::consumer()->usingDeserializer($deserializer);

```

--------------------------------

### Test Kafka Consumer with Mocking in PHP

Source: https://laravelkafka.com/docs/v2.0/testing/mocking-your-kafka-consumer

This example shows how to mock the Laravel Kafka facade to test consumer logic. It sets up expected messages using `shouldReceiveMessages` and then runs the consumer, allowing for assertions on the processed data. This approach isolates the consumer logic for unit testing.

```PHP
public function test_post_is_marked_as_published()
{
    // First, you use the fake method:
    \Junges\Kafka\Facades\Kafka::fake();

    // Then, tells Kafka what messages the consumer should receive:
    \Junges\Kafka\Facades\Kafka::shouldReceiveMessages([
        new \Junges\Kafka\Message\ConsumedMessage(
            topicName: 'mark-post-as-published-topic',
            partition: 0,
            headers: [],
            body: ['post_id' => 1],
            key: null,
            offset: 0,
            timestamp: 0
        ),
    ]);

    // Now, instantiate your consumer and start consuming messages. It will consume only the messages
    // specified in `shouldReceiveMessages` method:
    $consumer = \Junges\Kafka\Facades\Kafka::consumer(['mark-post-as-published-topic'])
        ->withHandler(function (\Junges\Kafka\Contracts\ConsumerMessage $message) use (&$posts) {
            $post = Post::find($message->getBody()['post_id']);

            $post->update(['published_at' => now()->format("Y-m-d H:i:s")]);

            return 0;

        })->build();

    $consumer->consume();

    // Now, you can test if the post published_at field is not empty, or anything else you want to test:

    $this->assertNotNull($post->refresh()->published_at);
}
```

--------------------------------

### Subscribe to a Single Kafka Topic

Source: https://laravelkafka.com/docs/v2.0/consuming-messages/subscribing-to-kafka-topics

Demonstrates the basic usage of subscribing a Kafka consumer to a single topic. This method is used to start consuming messages from a specified Kafka topic.

```PHP
use Junges\Kafka\Facades\Kafka;

$consumer = Kafka::consumer()->subscribe('topic');
```

--------------------------------

### Consume Kafka Messages

Source: https://laravelkafka.com/docs/v2.0/consuming-messages/consuming-messages

The `consume` method is called on a Kafka consumer instance to start processing incoming messages from Kafka topics. This is the primary method for message consumption.

```php
$consumer->consume();
```

--------------------------------

### Get Fresh Kafka Producer Instance

Source: https://laravelkafka.com/docs/v2.0/producing-messages/producing-messages

Retrieves a fresh Kafka Manager instance, allowing the creation of a new producer builder. Useful when the async producer's stored builder needs to be reset or a new instance is required.

```php
use Junges\Kafka\Facades\Kafka;

Kafka::fresh()
    ->asyncPublish('broker')
    ->onTopic('topic-name');
```

--------------------------------

### Assert Published On Times in PHP

Source: https://laravelkafka.com/docs/v2.0/testing/assert-published-on-times

This example demonstrates how to use the `assertPublishedOnTimes` method from Laravel Kafka to verify that a specific number of messages were published to a Kafka topic. It utilizes PHPUnit for testing.

```PHP
use PHPUnit\Framework\TestCase;
use Junges\Kafka\Facades\Kafka;
use Junges\Kafka\Message\Message;

class MyTest extends TestCase
{
    public function testWithSpecificTopic()
    {
        Kafka::fake();

        Kafka::publish('broker')
            ->onTopic('topic')
            ->withHeaders(['key' => 'value'])
            ->withBodyKey('key', 'value');

        Kafka::publish('broker')
            ->onTopic('topic')
            ->withHeaders(['key' => 'value'])
            ->withBodyKey('key', 'value');

        Kafka::assertPublishedOnTimes('some-kafka-topic', 2);
    }
}
```

--------------------------------

### Assert Published On Specific Topic

Source: https://laravelkafka.com/docs/v2.0/testing/assert-published-on

Verifies that a message was published to a specific Kafka topic. This example shows the basic usage without a callback for message content assertion.

```PHP
use PHPUnit\Framework\TestCase;
use Junges\Kafka\Facades\Kafka;

class MyTest extends TestCase
{
    public function testWithSpecificTopic()
    {
        Kafka::fake();

        $producer = Kafka::publish('broker')
            ->onTopic('some-kafka-topic')
            ->withHeaders(['key' => 'value'])
            ->withBodyKey('key', 'value');

        $producer->send();

        // Assert that a message was published to 'some-kafka-topic'
        Kafka::assertPublishedOn('some-kafka-topic', $producer->getMessage());

        // Alternatively, assert without providing the specific message object
        // Kafka::assertPublishedOn('some-kafka-topic');

    }
}
```

--------------------------------

### Assert published times

Source: https://laravelkafka.com/docs/v2.0/testing/assert-published-times

Use the `assertPublishedTimes` method to verify the number of messages published to Kafka. This example shows how to mock Kafka interactions with `Kafka::fake()` and then assert that exactly two messages were published.

```PHP
use PHPUnit\Framework\TestCase;
use Junges\Kafka\Facades\Kafka;
use Junges\Kafka\Message\Message;

class MyTest extends TestCase
{
    public function testWithSpecificTopic()
    {
        Kafka::fake();

        Kafka::publish('broker')
            ->onTopic('topic')
            ->withHeaders(['key' => 'value'])
            ->withBodyKey('key', 'value');

        Kafka::publish('broker')
            ->onTopic('topic')
            ->withHeaders(['key' => 'value'])
            ->withBodyKey('key', 'value');

        Kafka::assertPublishedTimes(2);
    }
}
```

--------------------------------

### Configure Kafka Consumer Batching

Source: https://laravelkafka.com/docs/v2.0/consuming-messages/handling-message-batch

This example shows how to enable batching for a Kafka consumer, specifying a maximum batch size of 1000 messages and a release interval of 1500 milliseconds. It includes a handler function to process the collected batch of messages before consuming them.

```php
$consumer = \Junges\Kafka\Facades\Kafka::consumer()
    ->enableBatching()
    ->withBatchSizeLimit(1000)
    ->withBatchReleaseInterval(1500)
    ->withHandler(function (\Illuminate\Support\Collection $collection, \Junges\Kafka\Contracts\MessageConsumer $consumer) {
         // Handle batch
    })
    ->build();

$consumer->consume();
```

--------------------------------

### Run Callback on Consumer Stop (PHP)

Source: https://laravelkafka.com/docs/v2.0/advanced-usage/graceful-shutdown

Demonstrates how to register a callback function that executes when a Kafka consumer instance stops consuming messages. This is useful for performing cleanup or final actions. The functionality requires the Process Control Extension to be installed.

```PHP
use Junges\Kafka\Facades\Kafka;

$consumer = Kafka::consumer(['topic'])
    ->withConsumerGroupId('group')
    ->withHandler(new Handler)
    ->onStopConsuming(static function () {
        // Do something when the consumer stop consuming messages
    })
    ->build();

$consumer->consume();

```

--------------------------------

### Create Basic Kafka Consumer

Source: https://laravelkafka.com/docs/v2.0/consuming-messages/creating-consumer

Demonstrates the initial step to create a Kafka consumer instance using the Kafka facade. This consumer can then be further configured to subscribe to topics.

```PHP
use Junges\Kafka\Facades\Kafka;

$consumer = Kafka::consumer();
```

--------------------------------

### Subscribe to Kafka Topics with Regex

Source: https://laravelkafka.com/docs/v2.0/consuming-messages/using-regex-to-subscribe-to-kafka-topics

Demonstrates subscribing to Kafka topics using a regular expression pattern. The `subscribe` method accepts a regex string, allowing dynamic topic matching. Topics intended for regex matching should be prefixed with `^`.

```PHP
\Junges\Kafka\Facades\Kafka::consumer()
    ->subscribe('^myPfx_.*')
    ->withHandler(...)
```

--------------------------------

### Create Kafka Consumer with Topics and Group

Source: https://laravelkafka.com/docs/v2.0/consuming-messages/creating-consumer

Shows how to create a Kafka consumer and simultaneously configure it with specific topics, a broker address, and a consumer group ID. This allows for more targeted message consumption from the outset.

```PHP
use Junges\Kafka\Facades\Kafka;

$consumer = Kafka::consumer(['topic-1', 'topic-2'], 'group-id', 'broker');
```

--------------------------------

### Kafka Environment Variables

Source: https://laravelkafka.com/docs/v2.0/example-docker-compose

Configuration variables for Kafka and ksqlDB producers and consumers, controlling interceptors, connect URLs, and logging topics.

```yaml
KSQL_PRODUCER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor"
KSQL_CONSUMER_INTERCEPTOR_CLASSES: "io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor"
KSQL_KSQL_CONNECT_URL: "http://connect:8083"
KSQL_KSQL_LOGGING_PROCESSING_TOPIC_REPLICATION_FACTOR: 1
KSQL_KSQL_LOGGING_PROCESSING_TOPIC_AUTO_CREATE: 'true'
KSQL_KSQL_LOGGING_PROCESSING_STREAM_AUTO_CREATE: 'true'
```

--------------------------------

### Configure Kafka Consumer Options

Source: https://laravelkafka.com/docs/v2.0/consuming-messages/configuring-consumer-options

Demonstrates setting Kafka consumer configuration options using the `withOptions` method for multiple options or the `withOption` method for a single option.

```PHP
$consumer = \Junges\Kafka\Facades\Kafka::consumer()
    ->withOptions([
        'option-name' => 'option-value'
    ]);
// Or:
$consumer = \Junges\Kafka\Facades\Kafka::consumer()
    ->withOption('option-name', 'option-value');

```

--------------------------------

### Create and Consume Kafka Messages

Source: https://laravelkafka.com/docs/v2.0/introduction

This PHP code snippet demonstrates how to instantiate a Kafka consumer in a Laravel application. It configures the consumer with topics, brokers, auto-commit settings, and a custom message handler, then initiates message consumption.

```PHP
$consumer = Kafka::consumer(["my-topic"])
    ->withBrokers("localhost:8092")
    ->withAutoCommit()
    ->withHandler(new MessageHandler)
    ->build();

$consumer->consume();
```

--------------------------------

### Build Kafka Consumer

Source: https://laravelkafka.com/docs/v2.0/consuming-messages/consuming-messages

Builds a Kafka consumer instance using the Kafka facade. This step is necessary before initiating message consumption.

```PHP
$consumer = \Junges\Kafka\Facades\Kafka::consumer()->build();
```

--------------------------------

### ksql-datagen Docker Service

Source: https://laravelkafka.com/docs/v2.0/example-docker-compose

Configures the ksql-datagen service for generating Kafka data, including image, dependencies, command for readiness checks, and environment variables for Kafka connection.

```yaml
ksql-datagen:
  image: confluentinc/ksqldb-examples:7.3.2
  hostname: ksql-datagen
  container_name: ksql-datagen
  depends_on:
    - ksqldb-server
    - broker
    - schema-registry
    - connect
  command: "bash -c 'echo Waiting for Kafka to be ready... && \
                   cub kafka-ready -b broker:29092 1 40 && \
                   echo Waiting for Confluent Schema Registry to be ready... && \
                   cub sr-ready schema-registry 8081 40 && \
                   echo Waiting a few seconds for topic creation to finish... && \
                   sleep 11 && \
                   tail -f /dev/null'"
  environment:
    KSQL_CONFIG_DIR: "/etc/ksql"
    STREAMS_BOOTSTRAP_SERVERS: broker:29092
    STREAMS_SCHEMA_REGISTRY_HOST: schema-registry
    STREAMS_SCHEMA_REGISTRY_PORT: 8081
```

--------------------------------

### rest-proxy Docker Service

Source: https://laravelkafka.com/docs/v2.0/example-docker-compose

Sets up the Kafka REST Proxy service, defining its image, dependencies, ports, hostname, container name, and environment variables for Kafka and Schema Registry integration.

```yaml
rest-proxy:
  image: confluentinc/cp-kafka-rest:7.3.2
  depends_on:
    - broker
    - schema-registry
  ports:
    - 8082:8082
  hostname: rest-proxy
  container_name: rest-proxy
  environment:
    KAFKA_REST_HOST_NAME: rest-proxy
    KAFKA_REST_BOOTSTRAP_SERVERS: 'broker:29092'
    KAFKA_REST_LISTENERS: "http://0.0.0.0:8082"
    KAFKA_REST_SCHEMA_REGISTRY_URL: 'http://schema-registry:8081'
```

--------------------------------

### Assigning Consumers to Topic Partitions

Source: https://laravelkafka.com/docs/v2.0/consuming-messages/assigning-partitions

Demonstrates how to use the `assignPartitions` method on a `ConsumerBuilder` instance to direct a consumer to specific topic partitions. This method accepts an array of `\RdKafka\TopicPartition` objects, allowing for precise control over partition consumption.

```PHP
$partition = 1; // The partition number you want to assign

$consumer = \Junges\Kafka\Facades\Kafka::consumer()
    ->assignPartitions([
        new \RdKafka\TopicPartition('your-topic-name', $partition)
    ]);

```

```PHP
$consumer = \Junges\Kafka\Facades\Kafka::consumer()
    ->assignPartitions([
        new \RdKafka\TopicPartition('your-topic-name', 1),
        new \RdKafka\TopicPartition('your-topic-name', 2),
        new \RdKafka\TopicPartition('your-topic-name', 3)
    ]);

```

--------------------------------

### Publishing a Message to Kafka

Source: https://laravelkafka.com/docs/v2.0/producing-messages/publishing-to-kafka

Demonstrates how to use the Kafka facade to publish a single message to a specified Kafka topic. It covers setting the broker, topic, configuration options, Kafka key, and headers before sending the message. The `send` method is recommended for low-throughput systems as it flushes the producer after every message.

```PHP
use Junges\Kafka\Facades\Kafka;

/** @var \Junges\Kafka\Producers\Builder $producer */
$producer = Kafka::publish('broker')
    ->onTopic('topic')
    ->withConfigOptions(['key' => 'value'])
    ->withKafkaKey('kafka-key')
    ->withHeaders(['header-key' => 'header-value']);

$producer->send();

```

--------------------------------

### Define Global Kafka Producer Configuration with Macro

Source: https://laravelkafka.com/docs/v2.0/advanced-usage/setting-global-configuration

This snippet shows how to extend the `Junges\Kafka\Facades\Kafka` class using Laravel's macro functionality to define a reusable producer configuration. It allows setting custom topic and configuration options that can be invoked later.

```PHP
// In a service provider:

\Junges\Kafka\Facades\Kafka::macro('myProducer', function () {
    return $this->publish('broker')
        ->onTopic('my-awesome-topic')
        ->withConfigOption('key', 'value');
});
```

--------------------------------

### Configuring Kafka Producer

Source: https://laravelkafka.com/docs/v2.0/producing-messages/configuring-producers

Details how to configure Kafka producer options using the producer builder returned by the publish call. This includes setting various producer-specific configurations.

```APIDOC
ProducerBuilder:
  publish(topic: string, payload: mixed): ProducerBuilder
    - Initiates the process of publishing a message to a Kafka topic.
    - Parameters:
      - topic: The Kafka topic to publish the message to.
      - payload: The message payload to be sent.
    - Returns: A ProducerBuilder instance for further configuration.

ProducerBuilder Methods:
  - Methods available on the ProducerBuilder instance allow for detailed configuration of the Kafka producer options before publishing.
```

--------------------------------

### Configure onStopConsuming Callbacks

Source: https://laravelkafka.com/docs/v2.0/upgrade-guide

Callbacks for `onStopConsuming` must now be defined during consumer construction, rather than after calling the `build` method, as was the practice in v1.13.x.

```php
$consumer = Kafka::consumer(['topic'])
    ->withConsumerGroupId('group')
    ->withHandler(new Handler)
    ->onStopConsuming(static function () {
        // Do something when the consumer stop consuming messages
    })
    ->build()
```

--------------------------------

### Produce Batch Messages with Laravel Kafka

Source: https://laravelkafka.com/docs/v2.0/producing-messages/producing-message-batch-to-kafka

Demonstrates the creation and usage of message batches for producing multiple messages to Kafka efficiently. It involves instantiating Message and MessageBatch objects, pushing messages, and sending the batch via the Kafka producer. This method is faster than sending individual messages for high-throughput scenarios.

```PHP
use Junges\Kafka\Facades\Kafka;
use Junges\Kafka\Producers\MessageBatch;
use Junges\Kafka\Message\Message;

$message = new Message(
    headers: ['header-key' => 'header-value'],
    body: ['key' => 'value'],
    key: 'kafka key here',
    topicName: 'my_topic'
);

$messageBatch = new MessageBatch();
$messageBatch->push($message);
$messageBatch->push($message);
$messageBatch->push($message);
$messageBatch->push($message);

/** @var \Junges\Kafka\Producers\Builder $producer */
$producer = Kafka::publish('broker')
    ->onTopic('topic')
    ->withConfigOptions(['key' => 'value']);

$producer->sendBatch($messageBatch);

```

--------------------------------

### Set Kafka Configuration Options

Source: https://laravelkafka.com/docs/v2.0/producing-messages/configuring-producers

Demonstrates how to set individual Kafka configuration options or multiple options at once for the producer. Configuration options are based on librdkafka settings. Refer to the librdkafka documentation for a full list of available properties.

```php
use Junges\Kafka\Facades\Kafka;

Kafka::publish('broker')
    ->onTopic('topic')
    ->withConfigOption('property-name', 'property-value')
    ->withConfigOptions([
        'property-name' => 'property-value'
    ]);

```

--------------------------------

### Kafka Facade Testing Assertions

Source: https://laravelkafka.com/docs/v2.0/testing/fake

The Kafka facade provides methods to perform assertions over published messages, allowing tests to verify message production to Kafka topics.

```APIDOC
Kafka Facade Testing Methods:

assertPublished(callable $callback = null)
  - Asserts that messages were published to Kafka.
  - Parameters:
    - callable $callback: An optional callback function to perform further assertions on the published messages.
  - Returns: void

assertPublishedOn(string $topic, callable $callback = null)
  - Asserts that messages were published to a specific Kafka topic.
  - Parameters:
    - string $topic: The name of the Kafka topic to check for published messages.
    - callable $callback: An optional callback function to perform further assertions on the published messages.
  - Returns: void

assertNothingPublished()
  - Asserts that no messages were published to Kafka.
  - Parameters: None
  - Returns: void

Related Methods:
  - assertPublishedTimes(string $topic, int $times): Asserts the number of times messages were published to a specific topic.
  - assertPublishedOnTimes(string $topic, int $times): Asserts the number of times messages were published to a specific topic.
```

--------------------------------

### Implement Queueable Kafka Handler in PHP

Source: https://laravelkafka.com/docs/v2.0/consuming-messages/queueable-handlers

Demonstrates how to create a queueable handler for Kafka messages by implementing the `ShouldQueue` interface. This allows messages to be processed asynchronously via Laravel's queue system. The handler receives Kafka messages via its `__invoke` method.

```PHP
use Illuminate\Contracts\Queue\ShouldQueue;
use Junges\Kafka\Contracts\Handler as HandlerContract;
use Junges\Kafka\Contracts\KafkaConsumerMessage;

class Handler implements HandlerContract, ShouldQueue
{
    public function __invoke(KafkaConsumerMessage $message): void
    {
        // Handle the consumed message.
        // Example: Log the message payload
        // Log::info('Received Kafka message:', $message->getPayload());
    }
}
```

```PHP
use Illuminate\Contracts\Queue\ShouldQueue;
use Junges\Kafka\Contracts\Handler as HandlerContract;
use Junges\Kafka\Contracts\KafkaConsumerMessage;

class Handler implements HandlerContract, ShouldQueue
{
    public function __invoke(KafkaConsumerMessage $message): void
    {
        // Handle the consumed message.
    }

    public function onConnection(): string
    {
        return 'sqs'; // Specify your queue connection, e.g., 'redis', 'sqs'
    }

    public function onQueue(): string
    {
        return 'kafka-handlers'; // Specify your queue name
    }
}
```

--------------------------------

### Configure and Use AVRO Serializer in Laravel Kafka

Source: https://laravelkafka.com/docs/v2.0/producing-messages/custom-serializers

This snippet demonstrates setting up the AVRO serializer for Laravel Kafka. It includes configuring the schema registry, mapping topic schemas, and creating a serializer instance for publishing messages.

```PHP
use FlixTech\AvroSerializer\Objects\RecordSerializer;
use FlixTech\SchemaRegistryApi\Registry\CachedRegistry;
use FlixTech\SchemaRegistryApi\Registry\BlockingRegistry;
use FlixTech\SchemaRegistryApi\Registry\PromisingRegistry;
use FlixTech\SchemaRegistryApi\Registry\Cache\AvroObjectCacheAdapter;
use GuzzleHttp\Client;

$cachedRegistry = new CachedRegistry(
    new BlockingRegistry(
        new PromisingRegistry(
            new Client(['base_uri' => 'kafka-schema-registry:9081'])
        )
    ),
    new AvroObjectCacheAdapter()
);

$registry = new AvroSchemaRegistry($cachedRegistry);
$recordSerializer = new RecordSerializer($cachedRegistry);

//if no version is defined, latest version will be used
//if no schema definition is defined, the appropriate version will be fetched form the registry
$registry->addBodySchemaMappingForTopic(
    'test-topic',
    new \Junges\Kafka\Message\KafkaAvroSchema('bodySchemaName' /*, int $version, AvroSchema $definition */)
);
$registry->addKeySchemaMappingForTopic(
    'test-topic',
    new \Junges\Kafka\Message\KafkaAvroSchema('keySchemaName' /*, int $version, AvroSchema $definition */)
);

$serializer = new \Junges\Kafka\Message\Serializers\AvroSerializer($registry, $recordSerializer /*, AvroEncoderInterface::ENCODE_BODY */);

$producer = \Junges\Kafka\Facades\Kafka::publish('broker')->onTopic('topic')->usingSerializer($serializer);
```

--------------------------------

### Subscribe to Multiple Kafka Topics

Source: https://laravelkafka.com/docs/v2.0/consuming-messages/subscribing-to-kafka-topics

Illustrates how to subscribe a Kafka consumer to multiple topics concurrently. This can be achieved by passing topic names as individual arguments or by providing an array of topic names to the `subscribe` method.

```PHP
use Junges\Kafka\Facades\Kafka;

$consumer = Kafka::consumer()->subscribe('topic-1', 'topic-2', 'topic-n');

// Or, using array:
$consumer = Kafka::consumer()->subscribe([
    'topic-1',
    'topic-2',
    'topic-n'
]);
```

--------------------------------

### Async Publish Kafka Message

Source: https://laravelkafka.com/docs/v2.0/producing-messages/producing-messages

Publishes messages asynchronously to Kafka. The async producer is a singleton that flushes messages when the application shuts down, reducing overhead for high-volume sending.

```php
use JungesKafka\Facades\Kafka;

Kafka::asyncPublish('broker')->onTopic('topic-name');
```

--------------------------------

### Define Reusable Kafka Producer Macro

Source: https://laravelkafka.com/docs/v2.0/advanced-usage/sending-multiple-messages-with-the-same-producer

This PHP code snippet demonstrates defining a custom macro for the Laravel Kafka facade. The `myProducer` macro configures a Kafka producer with a specific broker, topic, and configuration options, enabling reuse for sending multiple messages without re-specifying these details.

```PHP
// In a service provider:

\Junges\Kafka\Facades\Kafka::macro('myProducer', function () {
    return $this->publish('broker')
        ->onTopic('my-awesome-topic')
        ->withConfigOption('key', 'value');
});

```

--------------------------------

### Configure Message Body with withMessage

Source: https://laravelkafka.com/docs/v2.0/producing-messages/configuring-message-payload

The `withMessage` method allows you to set the entire message payload using an instance of `Junges\Kafka\Message\Message`.

```PHP
use Junges\Kafka\Facades\Kafka;
use Junges\Kafka\Message\Message;

$message = new Message(
    headers: ['header-key' => 'header-value'],
    body: ['key' => 'value'],
    key: 'kafka key here'
);

Kafka::publish('broker')->onTopic('topic')->withMessage($message);
```

--------------------------------

### Publish Message on Topic

Source: https://laravelkafka.com/docs/v2.0/upgrade-guide

The `publishOn` method has been renamed to `publish` and no longer accepts a `$topics` parameter. Messages must now be published by chaining `onTopic` to specify the target topic.

```php
\Junges\Kafka\Facades\Kafka::publish('broker')->onTopic('topic-name')
```

--------------------------------

### Add Middleware to Kafka Consumer

Source: https://laravelkafka.com/docs/v2.0/advanced-usage/middlewares

Demonstrates how to add a middleware to a Laravel Kafka consumer using the `withMiddleware` method. The middleware is a callable that receives the message and the next handler, allowing for custom message processing and execution in a defined order.

```PHP
$consumer = \Junges\Kafka\Facades\Kafka::consumer()
    ->withMiddleware(function(\Junges\Kafka\Message\ConsumedMessage $message, callable $next) {
        // Perform some work here
        return $next($message);
    });
```

--------------------------------

### Set Kafka Key with withKafkaKey

Source: https://laravelkafka.com/docs/v2.0/producing-messages/configuring-message-payload

The `withKafkaKey` method assigns a key to the message, which Kafka uses to determine the partition for message appending.

```PHP
use Junges\Kafka\Facades\Kafka;

Kafka::publish('broker')->onTopic('topic')->withKafkaKey('your-kafka-key');
```

--------------------------------

### Publish Message with Headers in Laravel Kafka

Source: https://laravelkafka.com/docs/v2.0/producing-messages/configuring-message-payload

Demonstrates how to publish a message to Kafka with custom headers using the Laravel Kafka package. It utilizes the `Kafka::publish` facade and the `withHeaders` method to attach key-value pairs to the message.

```PHP
use Junges\Kafka\Facades\Kafka;

Kafka::publish('broker')
    ->onTopic('topic')
    ->withHeaders([
        'header-key' => 'header-value'
    ])


```

--------------------------------

### Configure SASL Authentication with Laravel Kafka

Source: https://laravelkafka.com/docs/v2.0/advanced-usage/sasl-authentication

Configures SASL authentication for Kafka producers and consumers using the `withSasl` method. This method takes username, password, and the authentication mechanism as parameters. The security protocol defaults to `SASL_PLAINTEXT` if not explicitly set.

```PHP
$consumer = \Junges\Kafka\Facades\Kafka::consumer()
    ->withSasl(
        password: 'password',
        username: 'username',
        mechanisms: 'authentication mechanism'
    );
```

```PHP
$consumer = \Junges\Kafka\Facades\Kafka::consumer()
    ->withSasl(
        password: 'password',
        username: 'username',
        mechanisms: 'authentication mechanism',
        securityProtocol: 'SASL_SSL',
    );
```

--------------------------------

### Publish Kafka Message

Source: https://laravelkafka.com/docs/v2.0/producing-messages/producing-messages

Publishes a message to a specified Kafka topic using the synchronous producer. Returns a ProducerBuilder instance for further configuration.

```php
use Junges\Kafka\Facades\Kafka;

Kafka::publish('broker')->onTopic('topic-name');
```

--------------------------------

### Custom Deserializer Implementation

Source: https://laravelkafka.com/docs/v2.0/consuming-messages/custom-deserializers

To create a custom deserializer, implement the `\Junges\Kafka\Contracts\MessageDeserializer` contract. This requires defining the `deserialize` method. The custom deserializer must use the same algorithm as the serializer used during message production.

```PHP
namespace App\Kafka\Deserializers;

use Junges\Kafka\Contracts\MessageDeserializer;

class MyCustomDeserializer implements MessageDeserializer
{
    public function deserialize(string $data): mixed
    {
        // Your custom deserialization logic here
        return json_decode($data, true);
    }
}
```

--------------------------------

### Configure TLS Authentication for Kafka Consumer

Source: https://laravelkafka.com/docs/v2.0/advanced-usage/sasl-authentication

This snippet demonstrates how to configure TLS authentication options for a Kafka consumer using the Laravel Kafka library. It specifies the locations for CA certificate, client certificate, client key, and the endpoint identification algorithm.

```PHP
$consumer = \Junges\Kafka\Facades\Kafka::consumer()
    ->withOptions([
        'ssl.ca.location' => '/some/location/kafka.crt',
        'ssl.certificate.location' => '/some/location/client.crt',
        'ssl.key.location' => '/some/location/client.key',
        'ssl.endpoint.identification.algorithm' => 'none'
    ]);

```

--------------------------------

### Handle Kafka Messages with Callback

Source: https://laravelkafka.com/docs/v2.0/consuming-messages/message-handlers

Demonstrates how to set a message handler using a PHP closure for a Laravel Kafka consumer. It shows how to access message details within the callback function.

```PHP
$consumer = \Junges\Kafka\Facades\Kafka::consumer();

// Using callback:
$consumer->withHandler(function(\Junges\Kafka\Contracts\ConsumerMessage $message, \Junges\Kafka\Contracts\MessageConsumer $consumer) {
    // Handle your message here
});
```