Repository: mateusjunges/laravel-kafka (php)

Issue:
Kafka transaction when having multiple pushs
Hii there,
First of all, i appreciate your contribution for the laravel community !! i wanted to know if there's a way or a mechanism to make a transaction when publishing multiple topics after saving in the database, 

`Kafka::publishOn('topic1')->withMessage($message)->send();
            Kafka::publishOn('topic2')->withMessage($message)->send();`

i want to wrap it with something like kafka::transaction, when there's for example a problem in topic2 for some reason (like a time out or any other expection) it will remove the message published in topic2 or to send another message automatically to cancel it.

Any advice ?
Thank you in advance.

Pull request:
Add the ability to use Kafka transactional producers
Closes #217


## Transactional producers

Transactions provide the guarantee of atomicity, consistency, isolation, and durability (ACID) for a group of messages that are published to Kafka. This means that either all messages within the transaction will be successfully written to Kafka or none of the messages will be written. This is a crucial feature when it comes to ensuring data consistency and avoiding any data loss.

When using Kafka with transactional producers, each producer MUST configure its own static and unique [`transactional.id`](https://kafka.apache.org/documentation/#producerconfigs_transactional.id). This configuration option is used to uniquely identify the same logical producer across process restarts.

To guarantee message ordering, a given producer can have at most one ongoing transaction, and they are executed serially. 

## Consuming messages from transactional producers

When consuming messages from transactional producers it is important to set the `isolation.level` property of your consumer to `read_committed`, which will prevent consumers from advancing to offsets which includes open transactions. Messages from aborted transactions are filtered out within the consumer, and are never observed by the client application.



Patch:
diff --git a/build/laravel-kernels/kernel.php.stub b/build/laravel-kernels/kernel.php.stub
index de7e866..fa2b2c4 100644
--- a/build/laravel-kernels/kernel.php.stub
+++ b/build/laravel-kernels/kernel.php.stub
@@ -13,7 +13,7 @@ class Kernel extends ConsoleKernel
      * @var array
      */
     protected $commands = [
-        \Junges\Kafka\Console\Commands\KafkaConsumerCommand::class
+        \Junges\Kafka\Console\Commands\ConsumerCommand::class
     ];
 
     /**
diff --git a/composer.json b/composer.json
index 179b97f..30def74 100644
--- a/composer.json
+++ b/composer.json
@@ -6,7 +6,7 @@
         "php": "^8.1|^8.2",
         "ext-rdkafka": "^5.0|^6.0",
         "monolog/monolog": "^3.4",
-        "flix-tech/avro-serde-php": "^2.0",
+        "mateusjunges/avro-serde-php": "^3.0",
         "illuminate/support": "^9.0|^10.0",
         "illuminate/contracts": "^9.0|^10.0"
     },
diff --git a/src/Concerns/ManagesTransactions.php b/src/Concerns/ManagesTransactions.php
new file mode 100644
index 0000000..a36c79c
--- /dev/null
+++ b/src/Concerns/ManagesTransactions.php
@@ -0,0 +1,80 @@
+<?php declare(strict_types=1);
+
+namespace Junges\Kafka\Concerns;
+
+use Junges\Kafka\Exceptions\Transactions\TransactionFatalErrorException;
+use Junges\Kafka\Exceptions\Transactions\TransactionShouldBeAbortedException;
+use Junges\Kafka\Exceptions\Transactions\TransactionShouldBeRetriedException;
+use RdKafka\KafkaErrorException;
+
+/** @mixin \Junges\Kafka\Producers\Producer */
+trait ManagesTransactions
+{
+    /**
+     * @throws \Junges\Kafka\Exceptions\Transactions\TransactionShouldBeRetriedException
+     * @throws \Junges\Kafka\Exceptions\Transactions\TransactionFatalErrorException
+     * @throws \Junges\Kafka\Exceptions\Transactions\TransactionShouldBeAbortedException
+     */
+    public function beginTransaction(int $timeoutInMilliseconds = 1000): void
+    {
+        try {
+            if (! $this->transactionInitialized) {
+                $this->producer->initTransactions($timeoutInMilliseconds);
+                $this->transactionInitialized = true;
+            }
+
+            $this->producer->beginTransaction();
+        } catch (KafkaErrorException $exception) {
+            $this->handleTransactionException($exception);
+        }
+    }
+
+    /**
+     * @throws \Junges\Kafka\Exceptions\Transactions\TransactionShouldBeRetriedException
+     * @throws \Junges\Kafka\Exceptions\Transactions\TransactionFatalErrorException
+     * @throws \Junges\Kafka\Exceptions\Transactions\TransactionShouldBeAbortedException
+     */
+    public function abortTransaction(int $timeoutInMilliseconds = 1000): void
+    {
+        try {
+            $this->producer->abortTransaction($timeoutInMilliseconds);
+        } catch (KafkaErrorException $exception) {
+            $this->handleTransactionException($exception);
+        }
+    }
+
+    /**
+     * @throws \Junges\Kafka\Exceptions\Transactions\TransactionShouldBeRetriedException
+     * @throws \Junges\Kafka\Exceptions\Transactions\TransactionFatalErrorException
+     * @throws \Junges\Kafka\Exceptions\Transactions\TransactionShouldBeAbortedException
+     */
+    public function commitTransaction(int $timeoutInMilliseconds = 1000): void
+    {
+        try {
+            $this->producer->commitTransaction($timeoutInMilliseconds);
+        } catch (KafkaErrorException $exception) {
+            $this->handleTransactionException($exception);
+        }
+    }
+
+
+    /**
+     * @throws \Junges\Kafka\Exceptions\Transactions\TransactionShouldBeRetriedException
+     * @throws \Junges\Kafka\Exceptions\Transactions\TransactionShouldBeAbortedException
+     * @throws \Junges\Kafka\Exceptions\Transactions\TransactionFatalErrorException
+     */
+    private function handleTransactionException(KafkaErrorException $exception): void
+    {
+        if ($exception->isRetriable() === true) {
+            throw TransactionShouldBeRetriedException::new($exception);
+        }
+
+        if ($exception->transactionRequiresAbort() === true) {
+            throw TransactionShouldBeAbortedException::new($exception);
+        }
+
+        $this->transactionInitialized = false;
+
+        throw TransactionFatalErrorException::new($exception);
+    }
+}
\ No newline at end of file
diff --git a/src/Contracts/MessageProducer.php b/src/Contracts/MessageProducer.php
index e1a63c5..fd15f3e 100644
--- a/src/Contracts/MessageProducer.php
+++ b/src/Contracts/MessageProducer.php
@@ -5,6 +5,8 @@ namespace Junges\Kafka\Contracts;
 use Junges\Kafka\Config\Sasl;
 use Junges\Kafka\Message\Message;
 use Junges\Kafka\Producers\MessageBatch;
+use Junges\Kafka\Producers\Producer;
+use Junges\Kafka\Support\Testing\Fakes\ProducerFake;
 
 /**
  * @internal
@@ -12,11 +14,15 @@ use Junges\Kafka\Producers\MessageBatch;
 interface MessageProducer extends InteractsWithConfigCallbacks
 {
     /** Return a new Junges\Commit\ProducerBuilder instance. */
-    public static function create(string $topic, string $broker = null): self;
+    public static function create(string $broker = null): self;
 
     /** Sets a specific config option. */
     public function withConfigOption(string $name, mixed $option): self;
 
+    public function withTransactionalId(string $transactionalId): self;
+
+    public function onTopic(string $topic): self;
+
     /** Set offset commit callback to use with consumer groups. */
     public function withOffsetCommitCb(callable $callback): self;
 
@@ -50,9 +56,6 @@ interface MessageProducer extends InteractsWithConfigCallbacks
     /** Enables or disable debug. */
     public function withDebugEnabled(bool $enabled = true): self;
 
-    /** Returns the topic where the message will be published. */
-    public function getTopic(): string;
-
     /**
      * Send the given message to Kakfa.
      *
@@ -60,6 +63,8 @@ interface MessageProducer extends InteractsWithConfigCallbacks
      */
     public function send(): bool;
 
+    public function build(): Producer|ProducerFake;
+
     /**
      * Send a message batch to Kafka.
      *
diff --git a/src/Contracts/MessagePublisher.php b/src/Contracts/MessagePublisher.php
index a61ba9d..9bf7872 100644
--- a/src/Contracts/MessagePublisher.php
+++ b/src/Contracts/MessagePublisher.php
@@ -5,5 +5,5 @@ namespace Junges\Kafka\Contracts;
 interface MessagePublisher
 {
     /** Creates a new ProducerBuilder instance, setting brokers and topic. */
-    public function publishOn(string $topic, string $broker = null): MessageProducer;
+    public function publish(?string $broker = null): MessageProducer;
 }
diff --git a/src/Contracts/Producer.php b/src/Contracts/Producer.php
new file mode 100644
index 0000000..98cc7f4
--- /dev/null
+++ b/src/Contracts/Producer.php
@@ -0,0 +1,45 @@
+<?php declare(strict_types=1);
+
+namespace Junges\Kafka\Contracts;
+
+use Junges\Kafka\Exceptions\CouldNotPublishMessage;
+use Junges\Kafka\Producers\MessageBatch;
+use RdKafka\KafkaErrorException;
+
+interface Producer
+{
+    /**
+     * Produce the specified message in the kafka topic.
+     *
+     * @return mixed
+     * @throws \Exception
+     */
+    public function produce(ProducerMessage $message): bool;
+
+    /**
+     * @throws CouldNotPublishMessage
+     * @throws \Junges\Kafka\Exceptions\CouldNotPublishMessageBatch
+     */
+    public function produceBatch(MessageBatch $messageBatch): int;
+
+    /**
+     * @throws \Junges\Kafka\Exceptions\Transactions\TransactionShouldBeRetriedException
+     * @throws \Junges\Kafka\Exceptions\Transactions\TransactionFatalErrorException
+     * @throws \Junges\Kafka\Exceptions\Transactions\TransactionShouldBeAbortedException
+     */
+    public function beginTransaction(int $timeoutInMilliseconds = 1000): void;
+
+    /**
+     * @throws \Junges\Kafka\Exceptions\Transactions\TransactionShouldBeRetriedException
+     * @throws \Junges\Kafka\Exceptions\Transactions\TransactionFatalErrorException
+     * @throws \Junges\Kafka\Exceptions\Transactions\TransactionShouldBeAbortedException
+     */
+    public function abortTransaction(int $timeoutInMilliseconds = 1000): void;
+
+    /**
+     * @throws \Junges\Kafka\Exceptions\Transactions\TransactionShouldBeRetriedException
+     * @throws \Junges\Kafka\Exceptions\Transactions\TransactionFatalErrorException
+     * @throws \Junges\Kafka\Exceptions\Transactions\TransactionShouldBeAbortedException
+     */
+    public function commitTransaction(int $timeoutInMilliseconds = 1000): void;
+}
\ No newline at end of file
diff --git a/src/Contracts/ProducerMessage.php b/src/Contracts/ProducerMessage.php
index 3c88735..2ebacb1 100644
--- a/src/Contracts/ProducerMessage.php
+++ b/src/Contracts/ProducerMessage.php
@@ -4,12 +4,14 @@ namespace Junges\Kafka\Contracts;
 
 interface ProducerMessage extends KafkaMessage
 {
-    public static function create(string $topicName, int $partition): ProducerMessage;
+    public static function create(string $topicName = null, int $partition = RD_KAFKA_PARTITION_UA): ProducerMessage;
 
     public function withKey(?string $key): ProducerMessage;
 
     public function withBody(mixed $body): ProducerMessage;
 
+    public function onTopic(string $topic): ProducerMessage;
+
     public function withHeaders(array $headers = []):  ProducerMessage;
 
     public function withHeader(string $key, string|int|float $value): ProducerMessage;
diff --git a/src/Exceptions/CouldNotPublishMessageBatch.php b/src/Exceptions/CouldNotPublishMessageBatch.php
new file mode 100644
index 0000000..05aef4d
--- /dev/null
+++ b/src/Exceptions/CouldNotPublishMessageBatch.php
@@ -0,0 +1,16 @@
+<?php declare(strict_types=1);
+
+namespace Junges\Kafka\Exceptions;
+
+final class CouldNotPublishMessageBatch extends LaravelKafkaException
+{
+    public static function invalidTopicName(string $topic): self
+    {
+        $message = match (true) {
+            $topic === '' => "The provided topic name [''] is invalid for the message batch. Try again with a valid topic name.",
+            default => 'The provided topic name is invalid for the message batch. Try again with a valid topic name.'
+        };
+
+        return new self($message);
+    }
+}
\ No newline at end of file
diff --git a/src/Exceptions/Transactions/TransactionFatalErrorException.php b/src/Exceptions/Transactions/TransactionFatalErrorException.php
new file mode 100644
index 0000000..334fcb8
--- /dev/null
+++ b/src/Exceptions/Transactions/TransactionFatalErrorException.php
@@ -0,0 +1,20 @@
+<?php declare(strict_types=1);
+
+namespace Junges\Kafka\Exceptions\Transactions;
+
+use Junges\Kafka\Exceptions\LaravelKafkaException;
+use RdKafka\KafkaErrorException;
+
+final class TransactionFatalErrorException extends LaravelKafkaException
+{
+    private const FATAL_EXCEPTION_MESSAGE = 'Transaction failed with a fatal error. You must create a new producer as this one can not be used anymore. [%s]';
+
+    public static function new(KafkaErrorException $baseException): self
+    {
+        return new self(
+            sprintf(self::FATAL_EXCEPTION_MESSAGE, $baseException->getMessage()),
+            $baseException->getCode(),
+            $baseException
+        );
+    }
+}
\ No newline at end of file
diff --git a/src/Exceptions/Transactions/TransactionShouldBeAbortedException.php b/src/Exceptions/Transactions/TransactionShouldBeAbortedException.php
new file mode 100644
index 0000000..5b63137
--- /dev/null
+++ b/src/Exceptions/Transactions/TransactionShouldBeAbortedException.php
@@ -0,0 +1,20 @@
+<?php declare(strict_types=1);
+
+namespace Junges\Kafka\Exceptions\Transactions;
+
+use Junges\Kafka\Exceptions\LaravelKafkaException;
+use RdKafka\KafkaErrorException;
+
+final class TransactionShouldBeAbortedException extends LaravelKafkaException
+{
+    private const ABORTABLE_EXCEPTION_MESSAGE = 'Transaction failed. You must abort your current transaction and start a new one. [%s]';
+
+    public static function new(KafkaErrorException $baseException): self
+    {
+        return new self(
+            sprintf(self::ABORTABLE_EXCEPTION_MESSAGE, $baseException->getMessage()),
+            $baseException->getCode(),
+            $baseException
+        );
+    }
+}
\ No newline at end of file
diff --git a/src/Exceptions/Transactions/TransactionShouldBeRetriedException.php b/src/Exceptions/Transactions/TransactionShouldBeRetriedException.php
new file mode 100644
index 0000000..24d4d92
--- /dev/null
+++ b/src/Exceptions/Transactions/TransactionShouldBeRetriedException.php
@@ -0,0 +1,20 @@
+<?php declare(strict_types=1);
+
+namespace Junges\Kafka\Exceptions\Transactions;
+
+use Junges\Kafka\Exceptions\LaravelKafkaException;
+use RdKafka\KafkaErrorException;
+
+final class TransactionShouldBeRetriedException extends LaravelKafkaException
+{
+    private const RETRIABLE_EXCEPTION_MESSAGE = 'This transaction failed, but can be retried. [%s]';
+
+    public static function new(KafkaErrorException $baseException): self
+    {
+        return new self(
+            sprintf(self::RETRIABLE_EXCEPTION_MESSAGE, $baseException->getMessage()),
+            $baseException->getCode(),
+            $baseException
+        );
+    }
+}
\ No newline at end of file
diff --git a/src/Facades/Kafka.php b/src/Facades/Kafka.php
index 578315c..202df90 100644
--- a/src/Facades/Kafka.php
+++ b/src/Facades/Kafka.php
@@ -7,7 +7,7 @@ use Junges\Kafka\Contracts\ProducerMessage;
 use Junges\Kafka\Support\Testing\Fakes\KafkaFake;
 
 /**
- * @method static \Junges\Kafka\Contracts\MessageProducer publishOn(string $topic, string $broker = null)
+ * @method static \Junges\Kafka\Contracts\MessageProducer publish(string $broker = null)
  * @method static \Junges\Kafka\Consumers\ConsumerBuilder createConsumer(array $topics = [], string $groupId = null, string $brokers = null)
  * @method static void assertPublished(ProducerMessage $expectedMessage = null, callable $callback = null)
  * @method static void assertPublishedTimes(int $times = 1, ProducerMessage $expectedMessage = null, callable $callback = null)
diff --git a/src/Kafka.php b/src/Kafka.php
index ead1f80..5ae55fa 100644
--- a/src/Kafka.php
+++ b/src/Kafka.php
@@ -14,10 +14,9 @@ class Kafka implements MessagePublisher, ConsumeMessagesFromKafka
     use Macroable;
 
     /** Creates a new ProducerBuilder instance, setting brokers and topic. */
-    public function publishOn(string $topic, string $broker = null): MessageProducer
+    public function publish(string $broker = null): MessageProducer
     {
         return new ProducerBuilder(
-            topic: $topic,
             broker: $broker ?? config('kafka.brokers')
         );
     }
diff --git a/src/Message/Message.php b/src/Message/Message.php
index 919fbb9..3236173 100644
--- a/src/Message/Message.php
+++ b/src/Message/Message.php
@@ -41,6 +41,12 @@ class Message extends AbstractMessage implements Arrayable, ProducerMessage
         return $this;
     }
 
+    public function onTopic(string $topic): self
+    {
+        $this->topicName = $topic;
+        return $this;
+    }
+
     /** Set the kafka message key. */
     public function withKey(?string $key): Message
     {
diff --git a/src/Producers/MessageBatch.php b/src/Producers/MessageBatch.php
index 11cfc4d..d5dde50 100644
--- a/src/Producers/MessageBatch.php
+++ b/src/Producers/MessageBatch.php
@@ -19,7 +19,8 @@ class MessageBatch
 
     private string $uuid;
 
-    #[Pure]
+    private string $topic = '';
+
     public function __construct()
     {
         $this->messages = new SplDoublyLinkedList();
@@ -32,6 +33,13 @@ class MessageBatch
         $this->messages->push($message);
     }
 
+    public function onTopic(string $topicName): self
+    {
+        $this->topic = $topicName;
+
+        return $this;
+    }
+
     /**
      * Returns all messages from batch before producing them to kafka
      *
@@ -46,4 +54,9 @@ class MessageBatch
     {
         return $this->uuid;
     }
+
+    public function getTopicName(): string
+    {
+        return $this->topic;
+    }
 }
diff --git a/src/Producers/Producer.php b/src/Producers/Producer.php
index 0c70063..9c5db56 100644
--- a/src/Producers/Producer.php
+++ b/src/Producers/Producer.php
@@ -4,29 +4,36 @@ namespace Junges\Kafka\Producers;
 
 use Illuminate\Contracts\Events\Dispatcher;
 use Illuminate\Support\Facades\App;
+use Junges\Kafka\Concerns\ManagesTransactions;
 use Junges\Kafka\Config\Config;
-use Junges\Kafka\Contracts\MessageSerializer;
+use Junges\Kafka\Contracts\Producer as ProducerContract;
 use Junges\Kafka\Contracts\ProducerMessage;
+use Junges\Kafka\Contracts\MessageSerializer;
 use Junges\Kafka\Events\BatchMessagePublished;
 use Junges\Kafka\Events\MessageBatchPublished;
 use Junges\Kafka\Events\MessagePublished;
 use Junges\Kafka\Events\PublishingMessage;
 use Junges\Kafka\Events\PublishingMessageBatch;
 use Junges\Kafka\Exceptions\CouldNotPublishMessage;
+use Junges\Kafka\Exceptions\CouldNotPublishMessageBatch;
+use Junges\Kafka\Message\Message;
 use RdKafka\Conf;
 use RdKafka\Producer as KafkaProducer;
 use RdKafka\ProducerTopic;
 use SplDoublyLinkedList;
 
-class Producer
+class Producer implements ProducerContract
 {
+    use ManagesTransactions;
+
     private readonly KafkaProducer $producer;
     private readonly Dispatcher $dispatcher;
 
+    public bool $transactionInitialized = false;
+
     public function __construct(
         private readonly Config $config,
-        private readonly string $topic,
-        private readonly MessageSerializer $serializer
+        private readonly MessageSerializer $serializer,
     ) {
         $this->producer = app(KafkaProducer::class, [
             'conf' => $this->setConf($this->config->getProducerOptions()),
@@ -35,7 +42,7 @@ class Producer
     }
 
     /** Set the Kafka Configuration. */
-    public function setConf(array $options): Conf
+    private function setConf(array $options): Conf
     {
         $conf = new Conf();
 
@@ -50,17 +57,12 @@ class Producer
         return $conf;
     }
 
-    /**
-     * Produce the specified message in the kafka topic.
-     *
-     * @return mixed
-     * @throws \Exception
-     */
+    /** @inheritDoc */
     public function produce(ProducerMessage $message): bool
     {
         $this->dispatcher->dispatch(new PublishingMessage($message));
 
-        $topic = $this->producer->newTopic($this->topic);
+        $topic = $this->producer->newTopic($message->getTopicName());
 
         $message = clone $message;
 
@@ -73,10 +75,14 @@ class Producer
         return $this->flush();
     }
 
-    /** @throws CouldNotPublishMessage  */
+    /** @inheritDoc */
     public function produceBatch(MessageBatch $messageBatch): int
     {
-        $topic = $this->producer->newTopic($this->topic);
+        if ($messageBatch->getTopicName() === '') {
+            throw CouldNotPublishMessageBatch::invalidTopicName($messageBatch->getTopicName());
+        }
+
+        $topic = $this->producer->newTopic($messageBatch->getTopicName());
 
         $messagesIterator = $messageBatch->getMessages();
 
@@ -86,6 +92,8 @@ class Producer
 
         $this->dispatcher->dispatch(new PublishingMessageBatch($messageBatch));
         foreach ($messagesIterator as $message) {
+            assert($message instanceof Message);
+            $message->onTopic($messageBatch->getTopicName());
             $message = $this->serializer->serialize($message);
 
             $this->produceMessageBatch($topic, $message, $messageBatch->getBatchUuid());
diff --git a/src/Producers/ProducerBuilder.php b/src/Producers/ProducerBuilder.php
index 9ae5c12..e8baaab 100644
--- a/src/Producers/ProducerBuilder.php
+++ b/src/Producers/ProducerBuilder.php
@@ -16,29 +16,39 @@ class ProducerBuilder implements MessageProducer
     private array $options = [];
     private ProducerMessage $message;
     private MessageSerializer $serializer;
+    private Producer $producer;
+    private string $topic = '';
     private ?Sasl $saslConfig = null;
     private readonly string $broker;
+    private bool $isTransactionProducer = false;
+    private int $maxTransactionRetryAttempts = 5;
 
     public function __construct(
-        private readonly string $topic,
         ?string $broker = null,
     ) {
         /** @var ProducerMessage $message */
         $message = app(ProducerMessage::class);
-        $this->message = $message->create($topic);
+        $this->message = $message::create();
         $this->serializer = app(MessageSerializer::class);
         $this->broker = $broker ?? config('kafka.brokers');
     }
 
     /** Return a new Junges\Commit\ProducerBuilder instance. */
-    public static function create(string $topic, string $broker = null): self
+    public static function create(string $broker = null): self
     {
         return new ProducerBuilder(
-            topic: $topic,
             broker: $broker ?? config('kafka.brokers')
         );
     }
 
+    public function onTopic(string $topic): self
+    {
+        $this->topic = $topic;
+        $this->message->onTopic($topic);
+
+        return $this;
+    }
+
     /** Sets a specific config option. */
     public function withConfigOption(string $name, mixed $option): self
     {
@@ -47,6 +57,11 @@ class ProducerBuilder implements MessageProducer
         return $this;
     }
 
+    public function withTransactionalId(string $transactionalId): self
+    {
+        return $this->withConfigOption('transactional.id', $transactionalId);
+    }
+
     /** Sets configuration options. */
     public function withConfigOptions(array $options): self
     {
@@ -81,6 +96,14 @@ class ProducerBuilder implements MessageProducer
         return $this;
     }
 
+    public function transactional(int $maxRetryAttempts = 5): self
+    {
+        $this->isTransactionProducer = true;
+        $this->maxTransactionRetryAttempts = $maxRetryAttempts;
+
+        return $this;
+    }
+
     /** Set the message to be published.  */
     public function withMessage(ProducerMessage $message): self
     {
@@ -132,12 +155,6 @@ class ProducerBuilder implements MessageProducer
         return $this->withDebugEnabled(false);
     }
 
-    /** Returns the topic where the message will be published. */
-    public function getTopic(): string
-    {
-        return $this->topic;
-    }
-
     /**
      * Send the given message to Kakfa.
      *
@@ -159,14 +176,18 @@ class ProducerBuilder implements MessageProducer
     {
         $producer = $this->build();
 
+        if ($this->topic !== '' && $messageBatch->getTopicName() === '') {
+            $messageBatch->onTopic($this->topic);
+        }
+
         return $producer->produceBatch($messageBatch);
     }
 
-    private function build(): Producer
+    public function build(): Producer
     {
         $conf = new Config(
             broker: $this->broker,
-            topics: [$this->getTopic()],
+            topics: [],
             securityProtocol: $this->saslConfig?->getSecurityProtocol(),
             sasl: $this->saslConfig,
             customOptions: $this->options,
@@ -175,7 +196,6 @@ class ProducerBuilder implements MessageProducer
 
         return app(Producer::class, [
             'config' => $conf,
-            'topic' => $this->topic,
             'serializer' => $this->serializer,
         ]);
     }


Test patch:
diff --git a/src/Support/Testing/Fakes/KafkaFake.php b/src/Support/Testing/Fakes/KafkaFake.php
index 67ee6e5..66fd56a 100644
--- a/src/Support/Testing/Fakes/KafkaFake.php
+++ b/src/Support/Testing/Fakes/KafkaFake.php
@@ -22,9 +22,9 @@ class KafkaFake implements MessagePublisher
     }
 
     /** Publish a message in the specified broker/topic. */
-    public function publishOn(string $topic, ?string $broker = null): ProducerBuilderFake
+    public function publish(?string $broker = null): ProducerBuilderFake
     {
-        return $this->makeProducerBuilderFake($topic, $broker);
+        return $this->makeProducerBuilderFake($broker);
     }
 
     /** Return a ConsumerBuilder instance. */
@@ -91,9 +91,8 @@ class KafkaFake implements MessagePublisher
     {
         $count = $this->published($callback, $expectedMessage, $topic)->count();
 
-        PHPUnit::assertTrue(
-            condition: $count === $times,
-            message: "Kafka published {$count} messages instead of {$times}."
+        PHPUnit::assertSame(
+            $count, $times, "Kafka published {$count} messages instead of {$times}."
         );
     }
 
@@ -103,13 +102,12 @@ class KafkaFake implements MessagePublisher
         PHPUnit::assertEmpty($this->getPublishedMessages(), 'Messages were published unexpectedly.');
     }
 
-    private function makeProducerBuilderFake(string $topic = '', ?string $broker = null): ProducerBuilderFake
+    private function makeProducerBuilderFake(?string $broker = null): ProducerBuilderFake
     {
-        return (new ProducerBuilderFake(
-            topic: $topic,
-            broker: $broker
-        )
-        )->withProducerCallback(fn (Message $message) => $this->publishedMessages[] = $message);
+        return (new ProducerBuilderFake(broker: $broker))
+            ->withProducerCallback(
+                fn (Message $message) => $this->publishedMessages[] = $message
+            );
     }
 
     /*** Get all messages matching a truth-test callback. */
@@ -119,19 +117,22 @@ class KafkaFake implements MessagePublisher
             return collect();
         }
 
-        return collect($this->getPublishedMessages())->filter(function (Message $publishedMessage) use ($topic, $expectedMessage, $callback) {
-            if ($topic !== null && $publishedMessage->getTopicName() !== $topic) {
-                return false;
-            }
-            if ($callback !== null) {
-                return $callback($publishedMessage);
-            }
-            if ($expectedMessage !== null) {
-                return json_encode($publishedMessage->toArray(), JSON_THROW_ON_ERROR) === json_encode($expectedMessage->toArray(), JSON_THROW_ON_ERROR);
-            }
-
-            return true;
-        });
+        return collect($this->getPublishedMessages())
+            ->filter(function (Message $publishedMessage) use ($topic, $expectedMessage, $callback) {
+                if ($topic !== null && $publishedMessage->getTopicName() !== $topic) {
+                    return false;
+                }
+
+                if ($callback !== null) {
+                    return $callback($publishedMessage);
+                }
+
+                if ($expectedMessage !== null) {
+                    return json_encode($publishedMessage->toArray(), JSON_THROW_ON_ERROR) === json_encode($expectedMessage->toArray(), JSON_THROW_ON_ERROR);
+                }
+
+                return true;
+            });
     }
 
     /** Check if the producer has published messages. */
diff --git a/src/Support/Testing/Fakes/ProducerBuilderFake.php b/src/Support/Testing/Fakes/ProducerBuilderFake.php
index 73a55c0..223c27f 100644
--- a/src/Support/Testing/Fakes/ProducerBuilderFake.php
+++ b/src/Support/Testing/Fakes/ProducerBuilderFake.php
@@ -20,13 +20,13 @@ class ProducerBuilderFake implements MessageProducer
     private ProducerMessage $message;
     private MessageSerializer $serializer;
     private ?Sasl $saslConfig = null;
+    private string $topic = '';
     private ?Closure $producerCallback = null;
 
     public function __construct(
-        private readonly string $topic,
         private readonly ?string $broker = null,
     ) {
-        $this->message = new Message($topic);
+        $this->message = new Message();
 
         $conf = new Config(
             broker: '',
@@ -38,9 +38,16 @@ class ProducerBuilderFake implements MessageProducer
     }
 
     /** Return a new Junges\Commit\ProducerBuilder instance. */
-    public static function create(string $topic, string $broker = null): self
+    public static function create(string $broker = null): self
     {
-        return new ProducerBuilderFake($broker, $topic);
+        return new ProducerBuilderFake($broker);
+    }
+
+    public function onTopic(string $topic): self
+    {
+        $this->message->onTopic($topic);
+
+        return $this;
     }
 
     public function withProducerCallback(callable $callback): self
@@ -57,6 +64,11 @@ class ProducerBuilderFake implements MessageProducer
         return $this;
     }
 
+    public function withTransactionalId(string $transactionalId): self
+    {
+        return $this->withConfigOption('transactional.id', $transactionalId);
+    }
+
     /** Set config options */
     public function withConfigOptions(array $options): self
     {
@@ -115,12 +127,6 @@ class ProducerBuilderFake implements MessageProducer
         return $this;
     }
 
-    /** Get the kafka topic to be used. */
-    public function getTopic(): string
-    {
-        return $this->topic;
-    }
-
     public function getMessage(): Message
     {
         return $this->message;
@@ -149,10 +155,6 @@ class ProducerBuilderFake implements MessageProducer
     /** Send the message to the producer to be published on kafka. */
     public function send(): bool
     {
-        if ($this->getMessage()->getTopicName() === null) {
-            $this->message->setTopicName($this->getTopic());
-        }
-
         $producer = $this->build();
 
         return $producer->produce($this->getMessage());
@@ -163,6 +165,10 @@ class ProducerBuilderFake implements MessageProducer
     {
         $producer = $this->build();
 
+        if ($this->topic !== '' && $messageBatch->getTopicName() === '') {
+            $messageBatch->onTopic($this->topic);
+        }
+
         return $producer->produceBatch($messageBatch);
     }
 
@@ -170,7 +176,6 @@ class ProducerBuilderFake implements MessageProducer
     {
         $producerFake = app(ProducerFake::class, [
             'config' => $config,
-            'topic' => $this->getTopic(),
         ]);
 
         if ($this->producerCallback) {
@@ -181,11 +186,11 @@ class ProducerBuilderFake implements MessageProducer
     }
 
     /** Build the producer. */
-    private function build(): ProducerFake
+    public function build(): ProducerFake
     {
         $conf = new Config(
             broker: $this->broker ?? config('kafka.brokers'),
-            topics: [$this->getTopic()],
+            topics: [],
             sasl: $this->saslConfig,
             customOptions: $this->options,
             callbacks: $this->callbacks,
diff --git a/src/Support/Testing/Fakes/ProducerFake.php b/src/Support/Testing/Fakes/ProducerFake.php
index 9ba7de2..49228b2 100644
--- a/src/Support/Testing/Fakes/ProducerFake.php
+++ b/src/Support/Testing/Fakes/ProducerFake.php
@@ -3,19 +3,24 @@
 namespace Junges\Kafka\Support\Testing\Fakes;
 
 use Closure;
+use Junges\Kafka\Concerns\ManagesTransactions;
 use Junges\Kafka\Config\Config;
+use Junges\Kafka\Contracts\Producer;
+use Junges\Kafka\Contracts\ProducerMessage;
+use Junges\Kafka\Exceptions\CouldNotPublishMessageBatch;
 use Junges\Kafka\Message\Message;
 use Junges\Kafka\Producers\MessageBatch;
 use RdKafka\Conf;
 
-class ProducerFake
+class ProducerFake implements Producer
 {
+    use ManagesTransactions;
+
     private array $messages = [];
     private ?Closure $producerCallback = null;
 
     public function __construct(
         private readonly Config $config,
-        private readonly string $topic
     ) {
     }
 
@@ -31,7 +36,7 @@ class ProducerFake
         return $this;
     }
 
-    public function produce(Message $message): bool
+    public function produce(ProducerMessage $message): bool
     {
         if ($this->producerCallback !== null) {
             $callback = $this->producerCallback;
@@ -41,15 +46,20 @@ class ProducerFake
         return true;
     }
 
+    /** @throws \Junges\Kafka\Exceptions\CouldNotPublishMessageBatch */
     public function produceBatch(MessageBatch $messageBatch): int
     {
+        if ($messageBatch->getTopicName() === '') {
+            throw CouldNotPublishMessageBatch::invalidTopicName($messageBatch->getTopicName());
+        }
+
         $produced = 0;
         if ($this->producerCallback !== null) {
             $callback = $this->producerCallback;
 
             /** @var Message $message */
             foreach ($messageBatch->getMessages() as $message) {
-                $message->setTopicName($this->topic);
+                $message->onTopic($messageBatch->getTopicName());
                 $callback($message);
                 $produced++;
             }
diff --git a/tests/KafkaFakeTest.php b/tests/KafkaFakeTest.php
index df18a0c..83c5ce9 100644
--- a/tests/KafkaFakeTest.php
+++ b/tests/KafkaFakeTest.php
@@ -28,7 +28,8 @@ final class KafkaFakeTest extends LaravelKafkaTestCase
 
     public function testItStorePublishedMessagesOnArray(): void
     {
-        $producer = $this->fake->publishOn('topic')
+        $producer = $this->fake->publish()
+            ->onTopic('topic')
             ->withBodyKey('test', ['test'])
             ->withHeaders(['custom' => 'header'])
             ->withKafkaKey(Str::uuid()->toString());
@@ -46,7 +47,8 @@ final class KafkaFakeTest extends LaravelKafkaTestCase
             $this->assertThat($exception, new ExceptionMessage('The expected message was not published.'));
         }
 
-        $producer = $this->fake->publishOn('topic')
+        $producer = $this->fake->publish()
+            ->onTopic('topic')
             ->withBodyKey('test', ['test'])
             ->withHeaders(['custom' => 'header'])
             ->withKafkaKey(Str::uuid()->toString());
@@ -63,7 +65,8 @@ final class KafkaFakeTest extends LaravelKafkaTestCase
             $this->assertThat($exception, new ExceptionMessage('The expected message was not published.'));
         }
 
-        $producer = $this->fake->publishOn('topic')
+        $producer = $this->fake->publish()
+            ->onTopic('topic')
             ->withBodyKey('test', ['test'])
             ->withHeaders(['custom' => 'header'])
             ->withKafkaKey(Str::uuid()->toString());
@@ -85,7 +88,8 @@ final class KafkaFakeTest extends LaravelKafkaTestCase
 
     public function testItCanPerformAssertionsOnPublishedMessages(): void
     {
-        $producer = $this->fake->publishOn('topic')
+        $producer = $this->fake->publish()
+            ->onTopic('topic')
             ->withBodyKey('test', ['test'])
             ->withHeaders(['custom' => 'header'])
             ->withKafkaKey($uuid = Str::uuid()->toString());
@@ -114,7 +118,8 @@ final class KafkaFakeTest extends LaravelKafkaTestCase
 
     public function testAssertPublishedOn(): void
     {
-        $producer = $this->fake->publishOn('topic')
+        $producer = $this->fake->publish()
+            ->onTopic('topic')
             ->withBodyKey('test', ['test'])
             ->withHeaders(['custom' => 'header'])
             ->withKafkaKey(Str::uuid()->toString());
@@ -134,7 +139,8 @@ final class KafkaFakeTest extends LaravelKafkaTestCase
 
     public function testAssertPublishedOnTimes(): void
     {
-        $producer = $this->fake->publishOn('topic')
+        $producer = $this->fake->publish()
+            ->onTopic('topic')
             ->withBodyKey('test', ['test'])
             ->withHeaders(['custom' => 'header'])
             ->withKafkaKey(Str::uuid()->toString());
@@ -154,7 +160,8 @@ final class KafkaFakeTest extends LaravelKafkaTestCase
 
     public function testAssertPublishedOnTimesForBatchMessages(): void
     {
-        $producer = $this->fake->publishOn('batch-topic')
+        $producer = $this->fake->publish()
+            ->onTopic('batch-topic')
             ->withConfigOption('key', 'value');
 
         $message = new Message(
@@ -163,7 +170,7 @@ final class KafkaFakeTest extends LaravelKafkaTestCase
             key: 2
         );
 
-        $messageBatch = new MessageBatch();
+        $messageBatch = (new MessageBatch())->onTopic('batch-topic');
         $messageBatch->push($message);
         $messageBatch->push($message);
 
@@ -177,7 +184,8 @@ final class KafkaFakeTest extends LaravelKafkaTestCase
 
     public function testICanPerformAssertionsUsingAssertPublishedOn(): void
     {
-        $producer = $this->fake->publishOn('topic')
+        $producer = $this->fake->publish()
+            ->onTopic('topic')
             ->withBodyKey('test', ['test'])
             ->withHeaders(['custom' => 'header'])
             ->withKafkaKey($uuid = Str::uuid()->toString());
@@ -205,7 +213,7 @@ final class KafkaFakeTest extends LaravelKafkaTestCase
     {
         $this->fake->assertNothingPublished();
 
-        $this->fake->publishOn('topic', 'broker')->withMessage(new Message('foo'))->send();
+        $this->fake->publish('broker')->withMessage(new Message('foo'))->send();
 
         try {
             $this->fake->assertNothingPublished();
@@ -216,12 +224,12 @@ final class KafkaFakeTest extends LaravelKafkaTestCase
 
     public function testPublishMessageBatch(): void
     {
-        $messageBatch = new MessageBatch();
+        $messageBatch = (new MessageBatch())->onTopic('test');
         $messageBatch->push(new Message());
         $messageBatch->push(new Message());
         $messageBatch->push(new Message());
 
-        $producer = $this->fake->publishOn('topic')
+        $producer = $this->fake->publish()
             ->withBodyKey('test', ['test'])
             ->withHeaders(['custom' => 'header'])
             ->withKafkaKey(Str::uuid()->toString());
diff --git a/tests/KafkaTest.php b/tests/KafkaTest.php
index bba0953..9c5ef54 100644
--- a/tests/KafkaTest.php
+++ b/tests/KafkaTest.php
@@ -42,7 +42,8 @@ final class KafkaTest extends LaravelKafkaTestCase
 
         $this->app->bind(Producer::class, fn () => $mockedProducer);
 
-        $test = Kafka::publishOn('test-topic')
+        $test = Kafka::publish()
+            ->onTopic('test')
             ->withConfigOptions([
                 'metadata.broker.list' => 'broker',
             ])
@@ -74,7 +75,7 @@ final class KafkaTest extends LaravelKafkaTestCase
 
         $this->app->bind(Producer::class, fn () => $mockedProducer);
 
-        $producer = Kafka::publishOn('test-topic')
+        $producer = Kafka::publish()->onTopic('test-topic')
             ->withConfigOptions([
                 'metadata.broker.list' => 'broker',
             ])
@@ -107,7 +108,8 @@ final class KafkaTest extends LaravelKafkaTestCase
 
         Kafka::fake();
 
-        $test = Kafka::publishOn('test-topic')
+        $test = Kafka::publish()
+            ->onTopic('test-topic')
             ->withConfigOptions([
                 'metadata.broker.list' => 'broker',
             ])
@@ -139,9 +141,14 @@ final class KafkaTest extends LaravelKafkaTestCase
             return $mockedProducer;
         });
 
-        $message = Message::create()->withHeaders(['foo' => 'bar'])->withKey('message-key')->withBody(['foo' => 'bar']);
+        $message = Message::create()
+            ->withHeaders(['foo' => 'bar'])
+            ->onTopic('test')
+            ->withKey('message-key')
+            ->onTopic('test')
+            ->withBody(['foo' => 'bar']);
 
-        $test = Kafka::publishOn('test-topic')
+        $test = Kafka::publish()
             ->withConfigOptions([
                 'metadata.broker.list' => 'broker',
             ])
@@ -151,14 +158,15 @@ final class KafkaTest extends LaravelKafkaTestCase
 
         $this->assertTrue($test);
 
-        $test = Kafka::publishOn('test-topic')
+        $test = Kafka::publish()
             ->withConfigOptions([
                 'metadata.broker.list' => 'broker',
             ])
             ->withMessage(new Message(
+                topicName: 'test',
                 headers: ['foo' => 'bar'],
                 body: ['foo' => 'bar'],
-                key: 'message-key'
+                key: 'message-key',
             ))
             ->withDebugEnabled(false)
             ->send();
@@ -186,12 +194,13 @@ final class KafkaTest extends LaravelKafkaTestCase
         });
 
         /** @var ProducerBuilder $producer */
-        $producer = Kafka::publishOn('test-topic')
+        $producer = Kafka::publish()
             ->withConfigOptions([
                 'metadata.broker.list' => 'broker',
             ])
             ->withKafkaKey(Str::uuid()->toString())
             ->withBodyKey('test', ['test'])
+            ->onTopic('test')
             ->withHeaders(['custom' => 'header'])
             ->withDebugDisabled();
 
@@ -209,7 +218,7 @@ final class KafkaTest extends LaravelKafkaTestCase
 
     public function testICanUseCustomOptionsForProducerConfig(): void
     {
-        $producer = Kafka::publishOn('test-topic')
+        $producer = Kafka::publish()
             ->withConfigOptions($expectedOptions = [
                 'bootstrap.servers' => '[REMOTE_ADDRESS]',
                 'metadata.broker.list' => '[REMOTE_ADDRESS]',
@@ -265,7 +274,7 @@ final class KafkaTest extends LaravelKafkaTestCase
 
         $this->app->bind(Producer::class, fn () => $mockedProducer);
 
-        Kafka::publishOn('test')->withBodyKey('foo', 'bar')->send();
+        Kafka::publish()->onTopic('test')->withBodyKey('foo', 'bar')->send();
 
         Event::assertDispatched(CouldNotPublishMessageEvent::class, function (CouldNotPublishMessageEvent $event) use ($expectedMessage) {
             return $event->throwable instanceof CouldNotPublishMessage
@@ -276,10 +285,10 @@ final class KafkaTest extends LaravelKafkaTestCase
 
     public function testSendMessageBatch(): void
     {
-        $messageBatch = new MessageBatch();
-        $messageBatch->push(new Message());
-        $messageBatch->push(new Message());
-        $messageBatch->push(new Message());
+        $messageBatch = (new MessageBatch())->onTopic('test');
+        $messageBatch->push(new Message('test'));
+        $messageBatch->push(new Message('test'));
+        $messageBatch->push(new Message('test'));
 
         $expectedUuid = $messageBatch->getBatchUuid();
 
@@ -305,7 +314,7 @@ final class KafkaTest extends LaravelKafkaTestCase
 
         Event::fake();
 
-        Kafka::publishOn('test')->withBodyKey('foo', 'bar')->sendBatch($messageBatch);
+        Kafka::publish()->withBodyKey('foo', 'bar')->onTopic('test')->sendBatch($messageBatch);
 
         Event::assertDispatched(PublishingMessageBatch::class, function (PublishingMessageBatch $event) use ($messageBatch) {
             return $event->batch === $messageBatch;
@@ -320,15 +329,19 @@ final class KafkaTest extends LaravelKafkaTestCase
         });
     }
 
-    public function testMacro()
+    public function testMacro(): void
     {
         $sasl = new Sasl(username: 'username', password: 'password', mechanisms: 'mechanisms');
 
-        Kafka::macro('defaultProducer', function (string $topic) use ($sasl) {
-            return $this->publishOn($topic)->withSasl($sasl);
+        Kafka::macro('defaultProducer', function () {
+            return $this->publish()->withSasl(
+                username: 'username',
+                password: 'password',
+                mechanisms: 'mechanisms',
+            );
         });
 
-        $producer = Kafka::defaultProducer('test-topic');
+        $producer = Kafka::defaultProducer();
 
         $this->assertInstanceOf(ProducerBuilder::class, $producer);
         $this->assertEquals($sasl, $this->getPropertyWithReflection('saslConfig', $producer));
diff --git a/tests/Producers/ProducerTest.php b/tests/Producers/ProducerTest.php
index 5c432bb..17a0403 100644
--- a/tests/Producers/ProducerTest.php
+++ b/tests/Producers/ProducerTest.php
@@ -13,12 +13,13 @@ final class ProducerTest extends LaravelKafkaTestCase
     public function testItDoesNotDoubleSerializeMessageWhenUsingJsonSerializer(): void
     {
         $this->mockKafkaProducer();
-        $producer = new Producer(new Config('broker', ['test-topic']), 'test-topic', new JsonSerializer());
+        $producer = new Producer(new Config('broker', ['test-topic']), new JsonSerializer());
         $payload = ['key' => 'value'];
 
         $message = new Message(
             body: $payload,
         );
+        $message->onTopic('test-topic');
         $producer->produce($message);
         $producer->produce($message);