[Messenger] Added Kafka Transport Bridge

symfony · andythorne · Jul 21, 2023 · Jul 21, 2023 · Jul 21, 2023 · Jul 21, 2023
commit fd88c3ba39a22a83c3ee9500d6f2ab83f5921ac2
diff --git a/.github/workflows/integration-tests.yml b/.github/workflows/integration-tests.yml
@@ -99,7 +99,7 @@ jobs:
         ports:
           - 9092:9092
         env:
-          KAFKA_AUTO_CREATE_TOPICS_ENABLE: false
+          KAFKA_AUTO_CREATE_TOPICS_ENABLE: true
           KAFKA_CREATE_TOPICS: 'test-topic:1:1:compact'
           KAFKA_ADVERTISED_HOST_NAME: 127.0.0.1
           KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181'
@@ -135,7 +135,7 @@ jobs:
         uses: shivammathur/setup-php@v2
         with:
           coverage: "none"
-          extensions: "json,couchbase-3.2.2,memcached,mongodb-1.12.0,redis,rdkafka,xsl,ldap,relay"
+          extensions: "json,couchbase-3.2.2,memcached,mongodb-1.12.0,redis,rdkafka,xsl,ldap,relay,rdkafka"
           ini-values: date.timezone=UTC,memory_limit=-1,default_socket_timeout=10,session.gc_probability=0,apc.enable_cli=1,zend.assertions=1
           php-version: "${{ matrix.php }}"
           tools: pecl

diff --git a/.github/workflows/unit-tests.yml b/.github/workflows/unit-tests.yml
@@ -21,7 +21,7 @@ jobs:
     name: Unit Tests
 
     env:
-      extensions: amqp,apcu,igbinary,intl,mbstring,memcached,redis,relay
+      extensions: amqp,apcu,igbinary,intl,mbstring,memcached,redis,relay,rdkafka
 
     strategy:
       matrix:
@@ -43,6 +43,16 @@ jobs:
         with:
           fetch-depth: 2
 
+      - name: Install system dependencies
+        run: |
+          echo "::group::apt-get update"
+          sudo apt-get update
+          echo "::endgroup::"
+
+          echo "::group::install tools & libraries"
+          sudo apt-get install librdkafka-dev
+          echo "::endgroup::"
+
       - name: Setup PHP
         uses: shivammathur/setup-php@v2
         with:

diff --git a/src/Symfony/Component/Messenger/Bridge/Kafka/.gitattributes b/src/Symfony/Component/Messenger/Bridge/Kafka/.gitattributes
@@ -0,0 +1,4 @@
+/Tests export-ignore
+/phpunit.xml.dist export-ignore
+/.gitattributes export-ignore
+/.gitignore export-ignore
diff --git a/src/Symfony/Component/Messenger/Bridge/Kafka/.gitignore b/src/Symfony/Component/Messenger/Bridge/Kafka/.gitignore
@@ -0,0 +1,3 @@
+vendor/
+composer.lock
+phpunit.xml
diff --git a/src/Symfony/Component/Messenger/Bridge/Kafka/CHANGELOG.md b/src/Symfony/Component/Messenger/Bridge/Kafka/CHANGELOG.md
@@ -0,0 +1,7 @@
+CHANGELOG
+=========
+
+6.4
-6.4
+7.4
-6.4
+7.4
+---
+
+ * Introduced the Kafka bridge.
- * Introduced the Kafka bridge.
+ * Introduce the Kafka bridge.
- * Introduced the Kafka bridge.
+ * Introduce the Kafka bridge.
diff --git a/src/Symfony/Component/Messenger/Bridge/Kafka/Callback/LoggingErrorCallback.php b/src/Symfony/Component/Messenger/Bridge/Kafka/Callback/LoggingErrorCallback.php
@@ -0,0 +1,32 @@
+<?php
+
+declare(strict_types=1);
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Bridge\Kafka\Callback;
+
+use Psr\Log\LoggerInterface;
+use RdKafka\KafkaConsumer;
+
+final class LoggingErrorCallback
+{
+    public function __construct(
+        private readonly LoggerInterface $logger,
+    ) {
+    }
+
+    public function __invoke(KafkaConsumer $kafka, int $err, string $reason): void
+    {
+        $this->logger->error($reason, [
+            'error_code' => $err,
+        ]);
+    }
+}
diff --git a/src/Symfony/Component/Messenger/Bridge/Kafka/Callback/LoggingLogCallback.php b/src/Symfony/Component/Messenger/Bridge/Kafka/Callback/LoggingLogCallback.php
@@ -0,0 +1,35 @@
+<?php
+
+declare(strict_types=1);
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Bridge\Kafka\Callback;
+
+use Psr\Log\LoggerInterface;
+
+final class LoggingLogCallback
+{
+    public function __construct(
+        private readonly LoggerInterface $logger,
+    ) {
+    }
+
+    public function __invoke(object $kafka, int $level, string $facility, string $message): void
+    {
+        $this->logger->log(
+            $level,
+            $message,
+            [
+                'facility' => $facility,
+            ],
+        );
+    }
+}
diff --git a/src/Symfony/Component/Messenger/Bridge/Kafka/Callback/LoggingRebalanceCallback.php b/src/Symfony/Component/Messenger/Bridge/Kafka/Callback/LoggingRebalanceCallback.php
@@ -0,0 +1,109 @@
+<?php
+
+declare(strict_types=1);
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Bridge\Kafka\Callback;
+
+use Psr\Log\LoggerInterface;
+use RdKafka\KafkaConsumer;
+use RdKafka\TopicPartition;
+
+final class LoggingRebalanceCallback
+{
+    public function __construct(
+        private readonly LoggerInterface $logger,
+    ) {
+    }
+
+    /**
+     * @param list<TopicPartition>|null $topicPartitions
+     */
+    public function __invoke(KafkaConsumer $kafka, ?int $err, array $topicPartitions = null): void
+    {
+        $topicPartitions = $topicPartitions ?? [];
+
+        switch ($err) {
+            case \RD_KAFKA_RESP_ERR__ASSIGN_PARTITIONS:
+                foreach ($topicPartitions as $topicPartition) {
+                    $this->logger->info(
+                        sprintf(
+                            'Rebalancing %s %s %s as the assignment changed',
+                            $topicPartition->getTopic(),
+                            $topicPartition->getPartition(),
+                            $topicPartition->getOffset(),
+                        ),
+                        [
+                            'topic' => $topicPartition->getTopic(),
+                            'partition' => $topicPartition->getPartition(),
+                            'offset' => $topicPartition->getOffset(),
+                            'error_code' => $err,
+                        ],
+                    );
+                }
+                $kafka->assign($topicPartitions);
+                break;
+
+            case \RD_KAFKA_RESP_ERR__REVOKE_PARTITIONS:
+                foreach ($topicPartitions as $topicPartition) {
+                    $this->logger->info(
+                        sprintf(
+                            'Rebalancing %s %s %s as the assignment was revoked',
+                            $topicPartition->getTopic(),
+                            $topicPartition->getPartition(),
+                            $topicPartition->getOffset(),
+                        ),
+                        [
+                            'topic' => $topicPartition->getTopic(),
+                            'partition' => $topicPartition->getPartition(),
+                            'offset' => $topicPartition->getOffset(),
+                            'error_code' => $err,
+                        ],
+                    );
+                }
+                $kafka->assign(null);
+                break;
+
+            default:
+                if (\count($topicPartitions)) {
+                    foreach ($topicPartitions as $topicPartition) {
+                        $this->logger->error(
+                            sprintf(
+                                'Rebalancing %s %s %s due to error code %d',
+                                $topicPartition->getTopic(),
+                                $topicPartition->getPartition(),
+                                $topicPartition->getOffset(),
+                                $err,
+                            ),
+                            [
+                                'topic' => $topicPartition->getTopic(),
+                                'partition' => $topicPartition->getPartition(),
+                                'offset' => $topicPartition->getOffset(),
+                                'error_code' => $err,
+                            ],
+                        );
+                    }
+                } else {
+                    $this->logger->error(
+                        sprintf(
+                            'Rebalancing error code %d',
+                            $err,
+                        ),
+                        [
+                            'error_code' => $err,
+                        ]
+                    );
+                }
+                $kafka->assign(null);
+                break;
+        }
+    }
+}
diff --git a/src/Symfony/Component/Messenger/Bridge/Kafka/LICENSE b/src/Symfony/Component/Messenger/Bridge/Kafka/LICENSE
@@ -0,0 +1,19 @@
+Copyright (c) 2018-present Fabien Potencier
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is furnished
+to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.
diff --git a/src/Symfony/Component/Messenger/Bridge/Kafka/README.md b/src/Symfony/Component/Messenger/Bridge/Kafka/README.md
@@ -0,0 +1,12 @@
+Kafka Messenger
+===============
+
+Provides Kafka integration for Symfony Messenger.
+
+Resources
+---------
+
+ * [Contributing](https://symfony.com/doc/current/contributing/index.html)
+ * [Report issues](https://github.com/symfony/symfony/issues) and
+   [send Pull Requests](https://github.com/symfony/symfony/pulls)
+   in the [main Symfony repository](https://github.com/symfony/symfony)
diff --git a/src/Symfony/Component/Messenger/Bridge/Kafka/Stamp/KafkaMessageStamp.php b/src/Symfony/Component/Messenger/Bridge/Kafka/Stamp/KafkaMessageStamp.php
@@ -0,0 +1,26 @@
+<?php
+
+declare(strict_types=1);
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Bridge\Kafka\Stamp;
+
+use Symfony\Component\Messenger\Stamp\NonSendableStampInterface;
+
+final class KafkaMessageStamp implements NonSendableStampInterface
+{
+    public function __construct(
+        public int $partition,
+        public int $messageFlags,
+        public ?string $key,
+    ) {
+    }
+}
diff --git a/src/Symfony/Component/Messenger/Bridge/Kafka/Stamp/KafkaReceivedMessageStamp.php b/src/Symfony/Component/Messenger/Bridge/Kafka/Stamp/KafkaReceivedMessageStamp.php
@@ -0,0 +1,25 @@
+<?php
+
+declare(strict_types=1);
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Bridge\Kafka\Stamp;
+
+use RdKafka\Message;
+use Symfony\Component\Messenger\Stamp\NonSendableStampInterface;
+
+final class KafkaReceivedMessageStamp implements NonSendableStampInterface
+{
+    public function __construct(
+        public Message $message,
+    ) {
+    }
+}
diff --git a/src/Symfony/Component/Messenger/Bridge/Kafka/Tests/Callback/LoggingErrorCallbackTest.php b/src/Symfony/Component/Messenger/Bridge/Kafka/Tests/Callback/LoggingErrorCallbackTest.php
@@ -0,0 +1,38 @@
+<?php
+
+declare(strict_types=1);
+
+/*
+ * This file is part of the Symfony package.
+ *
+ * (c) Fabien Potencier <fabien@symfony.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace Symfony\Component\Messenger\Bridge\Kafka\Tests\Callback;
+
+use PHPUnit\Framework\TestCase;
+use Psr\Log\LoggerInterface;
+use RdKafka\KafkaConsumer;
+use Symfony\Component\Messenger\Bridge\Kafka\Callback\LoggingErrorCallback;
+
+/**
+ * @requires extension rdkafka
+ */
+final class LoggingErrorCallbackTest extends TestCase
+{
+    public function testInvoke(): void
+    {
+        $logger = $this->createMock(LoggerInterface::class);
+        $logger->expects(self::once())
+            ->method('error')
+            ->with('test error message', ['error_code' => 1]);
+
+        $consumer = $this->createMock(KafkaConsumer::class);
+
+        $callback = new LoggingErrorCallback($logger);
+        $callback($consumer, 1, 'test error message');
+    }
+}