Repository: mateusjunges/laravel-kafka (php)

Issue:
[BUG] Unable to set some configurations
**Describe the bug**
I think [this](https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md) is the official list of available configuration for brokers, topic, consumers and producers.  
Each configuration accepts a different kind of value: `string`, `integer`, `boolean`, `float`, `closure`.   
The `->withConfigOption` method accepts as value only `string`.   

**To Reproduce**
Steps to reproduce the behavior:
1. See [this](https://arnaud.le-blanc.net/php-rdkafka-doc/phpdoc/rdkafka-conf.seterrorcb.html) doc for example
2. Define this variable: `$callback = static function (mixed $kafka, int $errorCode, string $errorMsg) {
                throw new Exception($errorMsg, $errorCode);
            };`
3. Build a producer.
4. Use the following method: `->withConfigOption('error_cb', $callback)
5. Try to produce a message.
6. See the error: `TypeError : Junges\Kafka\Producers\ProducerBuilder::withConfigOption(): Argument #2 ($option) must be of type string, Closure given`

**Expected behavior**
For both producers and consumers, I expect one of the following alternatives (or any other else if better):
1. Change the type of the value of the option to `mixed`.
2. Make use of the methods available in this file `vendor/mateusjunges/laravel-kafka/src/Producers/Producer.php` on line 36 (see image attached). The `Conf()` class has some direct methods (`->setErrorCb()`, `->setDrMsgCb()`, `->setConsumeCb()`, `->setRebalanceCb()`, `->setOffsetCommitCb()`). Maybe add next to `->withConfigOption()` something like `->withErrorCb()`, `->withDrMsgCb()`, etc.

**Screenshots**
Image attached to show the available RdKafka\Conf methods.

**Additional context**
Please not that the first proposed solution has to be tested. The second solution shows direct methods that may mean that setting the callback through the related option key is not possible.

![Screenshot from 2022-08-24 13-13-51](https://user-images.githubusercontent.com/23011509/186405269-94345047-910d-4f0c-8c57-7f2827c9fdbe.png)


Pull request:
[v1.8.x] Add methods to configure config callbacks
This PR closes #135

Patch:
diff --git a/composer.json b/composer.json
index 2e9e4fe..4ca04c1 100644
--- a/composer.json
+++ b/composer.json
@@ -14,7 +14,8 @@
         "phpunit/phpunit": "^9.5",
         "orchestra/testbench": "^6.20",
         "predis/predis": "^1.1",
-        "friendsofphp/php-cs-fixer": "^3.0"
+        "friendsofphp/php-cs-fixer": "^3.0",
+        "kwn/php-rdkafka-stubs": "^2.2"
     },
     "autoload": {
         "psr-4": {
diff --git a/docs/consuming-messages/9-class-structure.md b/docs/consuming-messages/9-class-structure.md
new file mode 100644
index 0000000..7404c30
--- /dev/null
+++ b/docs/consuming-messages/9-class-structure.md
@@ -0,0 +1,72 @@
+---
+title: Class structure
+weight: 9
+---
+
+Consumer classes are very simple, and it is basically a Laravel Command class. To get started, let's take a look at an example consumer.
+
+```php
+<?php
+
+namespace App\Console\Commands\Consumers;
+
+use Illuminate\Console\Command;
+use Junges\Kafka\Facades\Kafka;
+use Junges\Kafka\Contracts\KafkaConsumerMessage;
+
+class MyTopicConsumer extends Command
+{
+    protected $signature = "consume:my-topic";
+
+    protected $description = "Consume Kafka messages from 'my-topic'."
+
+    public function handle()
+    {
+        $consumer = Kafka::createConsumer(['my-topic'])
+            ->withBrokers('localhost:8092')
+            ->withAutoCommit()
+            ->withHandler(function(KafkaConsumerMessage $message) {
+                // Handle your message here
+            })
+            ->build();
+            
+            $consumer->consume();
+    }
+}
+```
+
+Now, to keep this consumer process running permanently in the background, you should use a process monitor such as [supervisor](http://supervisord.org/) to ensure that the consumer does not stop running.
+
+## Supervisor configuration
+In production, you need a way to keep your consumer processes running. For this reason, you need to configure a process monitor that can detect when your consumer processes exit and automatically restart them. In addition, process monitors can allow you to specify how many consumer processes you would like to run concurrently. Supervisor is a process monitor commonly used in Linux environments and we will discuss how to configure it in the following documentation.
+
+### Installing supervisor
+To install supervisor on Ubuntu, you may use the following command:
+```bash
+sudo apt-get install supervisor
+```
+
+### Configuring supervisor
+Supervisor configuration files are typically stored in the `/etc/supervisor/conf.d` directory. Within this directory, you may create any number of configuration files that instruct supervisor how your processes should be monitored. For example, let's create a `my-topic-consumer.conf` file that starts and monitors our Consumer:
+
+```text
+[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
+```
+#### Starting Supervisor
+Onnce the configuration file has been created, you may update Supervisor configuration and start the processes using the following commands:
+
+```bash
+sudo supervisorctl reread
+
+sudo supervisorctl update
+
+sudo supervisorctl start my-topic-consumer:*
+```
diff --git a/src/Concerns/InteractsWithConfigCallbacks.php b/src/Concerns/InteractsWithConfigCallbacks.php
new file mode 100644
index 0000000..174ed7c
--- /dev/null
+++ b/src/Concerns/InteractsWithConfigCallbacks.php
@@ -0,0 +1,101 @@
+<?php
+
+namespace Junges\Kafka\Concerns;
+
+trait InteractsWithConfigCallbacks
+{
+    protected array $callbacks = [];
+
+    /**
+     * Set the configuration error callback.
+     *
+     * @param  callable  $callback
+     * @return $this
+     */
+    public function withErrorCb(callable $callback): self
+    {
+        $this->callbacks['setErrorCb'] = $callback;
+
+        return $this;
+    }
+
+    /**
+     * Sets the delivery report callback.
+     *
+     * @param  callable  $callback
+     * @return $this
+     */
+    public function withDrMsgCb(callable $callback): self
+    {
+        $this->callbacks['setDrMsgCb'] = $callback;
+
+        return $this;
+    }
+
+    /**
+     * Set consume callback to use with poll.
+     *
+     * @param  callable  $callback
+     * @return $this
+     */
+    public function withConsumeCb(callable $callback): self
+    {
+        $this->callbacks['setConsumeCb'] = $callback;
+
+        return $this;
+    }
+
+    /**
+     * Set the log callback.
+     *
+     * @param  callable  $callback
+     * @return $this
+     */
+    public function withLogCb(callable $callback): self
+    {
+        $this->callbacks['setLogCb'] = $callback;
+
+        return $this;
+    }
+
+    /**
+     * Set offset commit callback to use with consumer groups.
+     *
+     * @param  callable  $callback
+     * @return $this
+     */
+    public function withOffsetCommitCb(callable $callback): self
+    {
+        $this->callbacks['setOffsetCommitCb'] = $callback;
+
+        return $this;
+    }
+
+    /**
+     * Set rebalance callback for  use with coordinated consumer group balancing.
+     *
+     * @param  callable  $callback
+     * @return $this
+     */
+    public function withRebalanceCb(callable $callback): self
+    {
+        $this->callbacks['setRebalanceCb'] = $callback;
+
+        return $this;
+    }
+
+    /**
+     * Set statistics callback.
+     *
+     * @param  callable  $callback
+     * @return $this
+     */
+    public function withStatsCb(callable $callback): self
+    {
+        $this->callbacks['setStatsCb'] = $callback;
+
+        return $this;
+    }
+
+
+}
\ No newline at end of file
diff --git a/src/Config/Config.php b/src/Config/Config.php
index 5fc91e5..3b7a169 100644
--- a/src/Config/Config.php
+++ b/src/Config/Config.php
@@ -79,6 +79,7 @@ class Config
         ?HandlesBatchConfiguration $batchConfig = null,
         private bool               $stopAfterLastMessage = false,
         private int                $restartInterval = 1000,
+        private array              $callbacks = [],
     ) {
         $this->batchConfig = $batchConfig ?? new NullBatchConfig();
     }
@@ -165,6 +166,11 @@ class Config
         return $this->restartInterval;
     }
 
+    public function getConfigCallbacks(): array
+    {
+        return $this->callbacks;
+    }
+
     #[Pure]
     private function getSaslOptions(): array
     {
diff --git a/src/Consumers/Consumer.php b/src/Consumers/Consumer.php
index 4440bf6..5b4f6a6 100644
--- a/src/Consumers/Consumer.php
+++ b/src/Consumers/Consumer.php
@@ -165,6 +165,10 @@ class Consumer implements CanConsumeMessages
             $conf->set($key, $value);
         }
 
+        foreach ($this->config->getConfigCallbacks() as $method => $callback) {
+            $conf->{$method}($callback);
+        }
+
         return $conf;
     }
 
diff --git a/src/Consumers/ConsumerBuilder.php b/src/Consumers/ConsumerBuilder.php
index c289500..3fd8541 100644
--- a/src/Consumers/ConsumerBuilder.php
+++ b/src/Consumers/ConsumerBuilder.php
@@ -5,6 +5,7 @@ namespace Junges\Kafka\Consumers;
 use Closure;
 use InvalidArgumentException;
 use Junges\Kafka\Commit\Contracts\CommitterFactory;
+use Junges\Kafka\Concerns\InteractsWithConfigCallbacks;
 use Junges\Kafka\Config\BatchConfig;
 use Junges\Kafka\Config\Config;
 use Junges\Kafka\Config\NullBatchConfig;
@@ -18,6 +19,8 @@ use Junges\Kafka\Support\Timer;
 
 class ConsumerBuilder implements ConsumerBuilderContract
 {
+    use InteractsWithConfigCallbacks;
+
     protected array $topics;
     protected int $commit;
     protected ?string $groupId;
@@ -254,7 +257,7 @@ class ConsumerBuilder implements ConsumerBuilderContract
     /**
      * @inheritDoc
      */
-    public function withOption(string $name, string $value): self
+    public function withOption(string $name, mixed $value): self
     {
         $this->options[$name] = $value;
 
@@ -320,7 +323,8 @@ class ConsumerBuilder implements ConsumerBuilderContract
             autoCommit: $this->autoCommit,
             customOptions: $this->options,
             batchConfig: $this->getBatchConfig(),
-            stopAfterLastMessage: $this->stopAfterLastMessage
+            stopAfterLastMessage: $this->stopAfterLastMessage,
+            callbacks: $this->callbacks,
         );
 
         return new Consumer($config, $this->deserializer, $this->committerFactory);
diff --git a/src/Contracts/CanProduceMessages.php b/src/Contracts/CanProduceMessages.php
index 5c499df..189ce38 100644
--- a/src/Contracts/CanProduceMessages.php
+++ b/src/Contracts/CanProduceMessages.php
@@ -6,31 +6,140 @@ use Junges\Kafka\Config\Sasl;
 use Junges\Kafka\Message\Message;
 use Junges\Kafka\Producers\MessageBatch;
 
-interface CanProduceMessages
+/**
+ * @internal
+ */
+interface CanProduceMessages extends InteractsWithConfigCallbacks
 {
+    /**
+     * Return a new Junges\Commit\ProducerBuilder instance
+     * @param string $topic
+     * @param string|null $broker
+     * @return static
+     */
     public static function create(string $topic, string $broker = null): self;
 
-    public function withConfigOption(string $name, string $option): self;
+    /**
+     * Sets a specific config option.
+     *
+     * @param  string  $name
+     * @param  mixed  $option
+     * @return $this
+     */
+    public function withConfigOption(string $name, mixed $option): self;
 
+    /**
+     * Set offset commit callback to use with consumer groups.
+     *
+     * @param  callable  $callback
+     * @return $this
+     */
+    public function withOffsetCommitCb(callable $callback): self;
+
+    /**
+     * Set rebalance callback for  use with coordinated consumer group balancing.
+     *
+     * @param  callable  $callback
+     * @return $this
+     */
+    public function withRebalanceCb(callable $callback): self;
+
+    /**
+     * Set statistics callback.
+     *
+     * @param  callable  $callback
+     * @return $this
+     */
+    public function withStatsCb(callable $callback): self;
+
+
+    /**
+     * Sets configuration options.
+     *
+     * @param  array  $options
+     * @return $this
+     */
     public function withConfigOptions(array $options): self;
 
+    /**
+     * Set the message headers.
+     *
+     * @param array $headers
+     * @return $this
+     */
     public function withHeaders(array $headers = []): self;
 
+    /**
+     * Set the message key.
+     *
+     * @param string $key
+     * @return $this
+     */
     public function withKafkaKey(string $key): self;
 
+    /**
+     * Set a message array key.
+     *
+     * @param string $key
+     * @param mixed $message
+     * @return $this
+     */
     public function withBodyKey(string $key, mixed $message): self;
 
-    public function withMessage(Message $message): self;
+    /**
+     * Set the message to be published.
+     *
+     * @param  \Junges\Kafka\Contracts\KafkaProducerMessage  $message
+     * @return $this
+     */
+    public function withMessage(KafkaProducerMessage $message): self;
 
+    /**
+     * Set Sasl configuration.
+     *
+     * @param  \Junges\Kafka\Config\Sasl  $saslConfig
+     * @return $this
+     */
     public function withSasl(Sasl $saslConfig): self;
 
+    /**
+     * Specifies which serializer should be used.
+     *
+     * @param  \Junges\Kafka\Contracts\MessageSerializer  $serializer
+     * @return \Junges\Kafka\Contracts\CanProduceMessages
+     */
     public function usingSerializer(MessageSerializer $serializer): self;
 
+
+    /**
+     * Enables or disable debug.
+     *
+     * @param  bool  $enabled
+     * @return $this
+     */
     public function withDebugEnabled(bool $enabled = true): self;
 
+    /**
+     * Returns the topic where the message will be published.
+     *
+     * @return string
+     */
     public function getTopic(): string;
 
+    /**
+     * Send the given message to Kakfa.
+     *
+     * @throws \Exception
+     * @return bool
+     */
     public function send(): bool;
 
+    /**
+     * Send a message batch to Kafka.
+     *
+     * @param  \Junges\Kafka\Producers\MessageBatch  $messageBatch
+     * @throws \Junges\Kafka\Exceptions\CouldNotPublishMessage
+     * @return int
+     */
     public function sendBatch(MessageBatch $messageBatch): int;
 }
diff --git a/src/Contracts/ConsumerBuilder.php b/src/Contracts/ConsumerBuilder.php
index ce95289..202fe14 100644
--- a/src/Contracts/ConsumerBuilder.php
+++ b/src/Contracts/ConsumerBuilder.php
@@ -5,7 +5,10 @@ namespace Junges\Kafka\Contracts;
 use Junges\Kafka\Commit\Contracts\CommitterFactory;
 use Junges\Kafka\Config\Sasl;
 
-interface ConsumerBuilder
+/**
+ * @internal
+ */
+interface ConsumerBuilder extends InteractsWithConfigCallbacks
 {
     /**
      * Creates a new ConsumerBuilder instance.
@@ -144,10 +147,10 @@ interface ConsumerBuilder
      * Set a specific configuration option.
      *
      * @param string $name
-     * @param string $value
+     * @param mixed $value
      * @return \Junges\Kafka\Consumers\ConsumerBuilder
      */
-    public function withOption(string $name, string $value): self;
+    public function withOption(string $name, mixed $value): self;
 
     /**
      * Enables messages batching
diff --git a/src/Contracts/InteractsWithConfigCallbacks.php b/src/Contracts/InteractsWithConfigCallbacks.php
new file mode 100644
index 0000000..fe2962b
--- /dev/null
+++ b/src/Contracts/InteractsWithConfigCallbacks.php
@@ -0,0 +1,41 @@
+<?php
+
+namespace Junges\Kafka\Contracts;
+
+/**
+ * @internal
+ */
+interface InteractsWithConfigCallbacks
+{
+    /**
+     * Set the configuration error callback.
+     *
+     * @param  callable  $callback
+     * @return $this
+     */
+    public function withErrorCb(callable $callback): self;
+
+    /**
+     * Sets the delivery report callback.
+     *
+     * @param  callable  $callback
+     * @return $this
+     */
+    public function withDrMsgCb(callable $callback): self;
+
+    /**
+     * Set consume callback to use with poll.
+     *
+     * @param  callable  $callback
+     * @return $this
+     */
+    public function withConsumeCb(callable $callback): self;
+
+    /**
+     * Set the log callback.
+     *
+     * @param  callable  $callback
+     * @return $this
+     */
+    public function withLogCb(callable $callback): self;
+}
\ No newline at end of file
diff --git a/src/Contracts/KafkaConsumerMessage.php b/src/Contracts/KafkaConsumerMessage.php
index 1104e46..d168a77 100644
--- a/src/Contracts/KafkaConsumerMessage.php
+++ b/src/Contracts/KafkaConsumerMessage.php
@@ -2,6 +2,9 @@
 
 namespace Junges\Kafka\Contracts;
 
+/**
+ * @internal
+ */
 interface KafkaConsumerMessage extends KafkaMessage
 {
     public function getOffset(): ?int;
diff --git a/src/Contracts/KafkaMessage.php b/src/Contracts/KafkaMessage.php
index c223f7d..e065cf5 100644
--- a/src/Contracts/KafkaMessage.php
+++ b/src/Contracts/KafkaMessage.php
@@ -2,6 +2,9 @@
 
 namespace Junges\Kafka\Contracts;
 
+/**
+ * @internal
+ */
 interface KafkaMessage
 {
     public function getKey(): mixed;
diff --git a/src/Producers/Producer.php b/src/Producers/Producer.php
index 9e6bdee..1982927 100644
--- a/src/Producers/Producer.php
+++ b/src/Producers/Producer.php
@@ -39,6 +39,10 @@ class Producer
             $conf->set($key, $value);
         }
 
+        foreach ($this->config->getConfigCallbacks() as $method => $callback) {
+            $conf->{$method}($callback);
+        }
+
         return $conf;
     }
 
diff --git a/src/Producers/ProducerBuilder.php b/src/Producers/ProducerBuilder.php
index 4404593..4dbd654 100644
--- a/src/Producers/ProducerBuilder.php
+++ b/src/Producers/ProducerBuilder.php
@@ -2,15 +2,17 @@
 
 namespace Junges\Kafka\Producers;
 
+use Junges\Kafka\Concerns\InteractsWithConfigCallbacks;
 use Junges\Kafka\Config\Config;
 use Junges\Kafka\Config\Sasl;
 use Junges\Kafka\Contracts\CanProduceMessages;
 use Junges\Kafka\Contracts\KafkaProducerMessage;
 use Junges\Kafka\Contracts\MessageSerializer;
-use Junges\Kafka\Message\Message;
 
 class ProducerBuilder implements CanProduceMessages
 {
+    use InteractsWithConfigCallbacks;
+
     private array $options = [];
     private KafkaProducerMessage $message;
     private MessageSerializer $serializer;
@@ -42,13 +44,26 @@ class ProducerBuilder implements CanProduceMessages
         );
     }
 
-    public function withConfigOption(string $name, string $option): self
+    /**
+     * Sets a specific config option.
+     *
+     * @param  string  $name
+     * @param  mixed  $option
+     * @return $this
+     */
+    public function withConfigOption(string $name, mixed $option): self
     {
         $this->options[$name] = $option;
 
         return $this;
     }
 
+    /**
+     * Sets configuration options.
+     *
+     * @param  array  $options
+     * @return $this
+     */
     public function withConfigOptions(array $options): self
     {
         foreach ($options as $name => $value) {
@@ -60,6 +75,7 @@ class ProducerBuilder implements CanProduceMessages
 
     /**
      * Set the message headers.
+     *
      * @param array $headers
      * @return $this
      */
@@ -72,6 +88,7 @@ class ProducerBuilder implements CanProduceMessages
 
     /**
      * Set the message key.
+     *
      * @param string $key
      * @return $this
      */
@@ -84,9 +101,10 @@ class ProducerBuilder implements CanProduceMessages
 
     /**
      * Set a message array key.
+     *
      * @param string $key
      * @param mixed $message
-     * @return ProducerBuilder
+     * @return $this
      */
     public function withBodyKey(string $key, mixed $message): self
     {
@@ -95,6 +113,12 @@ class ProducerBuilder implements CanProduceMessages
         return $this;
     }
 
+    /**
+     * Set the message to be published.
+     *
+     * @param  \Junges\Kafka\Contracts\KafkaProducerMessage  $message
+     * @return $this
+     */
     public function withMessage(KafkaProducerMessage $message): self
     {
         $this->message = $message;
@@ -102,6 +126,12 @@ class ProducerBuilder implements CanProduceMessages
         return $this;
     }
 
+    /**
+     * Enables or disable debug.
+     *
+     * @param  bool  $enabled
+     * @return $this
+     */
     public function withDebugEnabled(bool $enabled = true): self
     {
         if ($enabled) {
@@ -118,6 +148,8 @@ class ProducerBuilder implements CanProduceMessages
     }
 
     /**
+     * Set Sasl configuration.
+     *
      * @param Sasl $saslConfig
      * @return $this
      */
@@ -128,6 +160,12 @@ class ProducerBuilder implements CanProduceMessages
         return $this;
     }
 
+    /**
+     * Specifies which serializer should be used.
+     *
+     * @param  \Junges\Kafka\Contracts\MessageSerializer  $serializer
+     * @return \Junges\Kafka\Contracts\CanProduceMessages
+     */
     public function usingSerializer(MessageSerializer $serializer): CanProduceMessages
     {
         $this->serializer = $serializer;
@@ -135,16 +173,32 @@ class ProducerBuilder implements CanProduceMessages
         return $this;
     }
 
+    /**
+     * Disables debug.
+     *
+     * @return $this
+     */
     public function withDebugDisabled(): self
     {
         return $this->withDebugEnabled(false);
     }
 
+    /**
+     * Returns the topic where the message will be published.
+     *
+     * @return string
+     */
     public function getTopic(): string
     {
         return $this->topic;
     }
 
+    /**
+     * Send the given message to Kakfa.
+     *
+     * @throws \Exception
+     * @return bool
+     */
     public function send(): bool
     {
         $producer = $this->build();
@@ -152,6 +206,13 @@ class ProducerBuilder implements CanProduceMessages
         return $producer->produce($this->message);
     }
 
+    /**
+     * Send a message batch to Kafka.
+     *
+     * @param  \Junges\Kafka\Producers\MessageBatch  $messageBatch
+     * @throws \Junges\Kafka\Exceptions\CouldNotPublishMessage
+     * @return int
+     */
     public function sendBatch(MessageBatch $messageBatch): int
     {
         $producer = $this->build();
@@ -167,6 +228,7 @@ class ProducerBuilder implements CanProduceMessages
             securityProtocol: $this->saslConfig?->getSecurityProtocol(),
             sasl: $this->saslConfig,
             customOptions: $this->options,
+            callbacks: $this->callbacks,
         );
 
         return app(Producer::class, [


Test patch:
diff --git a/docs/testing/7-mocking-your-kafka-consumer.md b/docs/testing/7-mocking-your-kafka-consumer.md
index 50c22b4..787513a 100644
--- a/docs/testing/7-mocking-your-kafka-consumer.md
+++ b/docs/testing/7-mocking-your-kafka-consumer.md
@@ -43,6 +43,8 @@ public function test_post_is_marked_as_published()
 
         })->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);
diff --git a/src/Support/Testing/Fakes/ConsumerBuilderFake.php b/src/Support/Testing/Fakes/ConsumerBuilderFake.php
index d6b8525..3a30236 100644
--- a/src/Support/Testing/Fakes/ConsumerBuilderFake.php
+++ b/src/Support/Testing/Fakes/ConsumerBuilderFake.php
@@ -65,6 +65,7 @@ class ConsumerBuilderFake extends ConsumerBuilder implements ConsumerBuilderCont
             customOptions: $this->options,
             batchConfig: $this->getBatchConfig(),
             stopAfterLastMessage: $this->stopAfterLastMessage,
+            callbacks: $this->callbacks,
         );
 
         return new ConsumerFake(
diff --git a/src/Support/Testing/Fakes/ProducerBuilderFake.php b/src/Support/Testing/Fakes/ProducerBuilderFake.php
index 59280f3..94f535b 100644
--- a/src/Support/Testing/Fakes/ProducerBuilderFake.php
+++ b/src/Support/Testing/Fakes/ProducerBuilderFake.php
@@ -3,6 +3,7 @@
 namespace Junges\Kafka\Support\Testing\Fakes;
 
 use Closure;
+use Junges\Kafka\Concerns\InteractsWithConfigCallbacks;
 use Junges\Kafka\Config\Config;
 use Junges\Kafka\Config\Sasl;
 use Junges\Kafka\Contracts\CanProduceMessages;
@@ -13,6 +14,8 @@ use Junges\Kafka\Producers\MessageBatch;
 
 class ProducerBuilderFake implements CanProduceMessages
 {
+    use InteractsWithConfigCallbacks;
+
     private array $options = [];
     private KafkaProducerMessage $message;
     private MessageSerializer $serializer;
@@ -52,7 +55,7 @@ class ProducerBuilderFake implements CanProduceMessages
         return $this;
     }
 
-    public function withConfigOption(string $name, string $option): self
+    public function withConfigOption(string $name, mixed $option): self
     {
         $this->options[$name] = $option;
 
@@ -101,7 +104,7 @@ class ProducerBuilderFake implements CanProduceMessages
      * Set a message array key.
      * @param string $key
      * @param mixed $message
-     * @return ProducerBuilderFake
+     * @return $this
      */
     public function withBodyKey(string $key, mixed $message): self
     {
@@ -113,10 +116,10 @@ class ProducerBuilderFake implements CanProduceMessages
     /**
      * Set the entire message.
      *
-     * @param Message $message
+     * @param \Junges\Kafka\Contracts\KafkaProducerMessage $message
      * @return $this
      */
-    public function withMessage(Message $message): self
+    public function withMessage(KafkaProducerMessage $message): self
     {
         $this->message = $message;
 
@@ -125,6 +128,7 @@ class ProducerBuilderFake implements CanProduceMessages
 
     /**
      * Enable or disable kafka debug.
+     *
      * @param bool $enabled
      * @return $this
      */
@@ -164,6 +168,12 @@ class ProducerBuilderFake implements CanProduceMessages
         return $this;
     }
 
+    /**
+     * Specifies which serializer should be used.
+     *
+     * @param  \Junges\Kafka\Contracts\MessageSerializer  $serializer
+     * @return \Junges\Kafka\Contracts\CanProduceMessages
+     */
     public function usingSerializer(MessageSerializer $serializer): CanProduceMessages
     {
         $this->serializer = $serializer;
@@ -173,6 +183,7 @@ class ProducerBuilderFake implements CanProduceMessages
 
     /**
      * Send the message to the producer to be published on kafka.
+     *
      * @return bool
      */
     public function send(): bool
@@ -186,6 +197,12 @@ class ProducerBuilderFake implements CanProduceMessages
         return $producer->produce($this->getMessage());
     }
 
+    /**
+     * Send a message batch to Kafka.
+     *
+     * @param  \Junges\Kafka\Producers\MessageBatch  $messageBatch
+     * @return int
+     */
     public function sendBatch(MessageBatch $messageBatch): int
     {
         $producer = $this->build();
@@ -218,7 +235,8 @@ class ProducerBuilderFake implements CanProduceMessages
             broker: $this->broker ?? config('kafka.brokers'),
             topics: [$this->getTopic()],
             sasl: $this->saslConfig,
-            customOptions: $this->options
+            customOptions: $this->options,
+            callbacks: $this->callbacks,
         );
 
         return $this->makeProducer($conf);