[Messenger] Kafka Transport Bridge #39712

KonstantinCodes · 2021-01-04T16:02:15Z

Q	A
Branch?	6.3
Bug fix?	no
New feature?	yes
Deprecations?	no
Tickets	Fix #35549
License	MIT
Doc PR	symfony/symfony-docs#15884

This PR integrates a simple Kafka Transport for Messenger.

src/Symfony/Component/Messenger/Bridge/Kafka/composer.json

src/Symfony/Component/Messenger/Bridge/Kafka/phpstan.neon

derrabus

Thank you for your contribution.

I wonder how difficult it might be to test against an actual Kafka instance in our integration test suite. What do you think?

src/Symfony/Component/Messenger/Bridge/Kafka/Tests/Transport/KafkaTransportFactoryTest.php

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaReceiver.php

KonstantinCodes · 2021-01-04T21:18:25Z

@derrabus thank you for your review! I'll get started with the fixes tomorrow.

Actual integration tests are possible. I have a test that writes and reads. I see there is a kafka instance running in travis, so that's cool!

Do you think we could install the current rdkafka on travis?

derrabus · 2021-01-04T21:28:35Z

Do you think we could install the current rdkafka on travis?

Our integration tests run on GitHub Actions. That extension should already be enabled there.

KonstantinCodes · 2021-01-04T21:38:13Z

Oh nice. What about travis though? This test run (Unit Tests) fails because of the missing rdkafka.

derrabus · 2021-01-04T22:04:35Z

There's a block Install extra PHP extensions in .travis.yaml. You can add the extension there.

jderusse

Thank for working on this.

Few comments from my side. I think you should remove some logs, and reduce the verbosity.

src/Symfony/Component/Messenger/Bridge/Kafka/Tests/Fixtures/TestMessage.php

jderusse · 2021-01-05T09:47:58Z

src/Symfony/Component/Messenger/Bridge/Kafka/Tests/Transport/KafkaTransportFactoryTest.php

+    public function testSupports()
+    {
+        static::assertTrue($this->factory->supports('kafka://my-local-kafka:9092', []));
+        static::assertTrue($this->factory->supports('kafka+ssl://my-staging-kafka:9093', []));


use querystring parameter sslmode=disable for consistency with sqs adapter and pgsql

The TLS config is done through settings passed in the conf array, and conforms to the underlying librdkafka format. The protocol in the DSN is only used to recognize this DSN as a Kafka-DSN.

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaMessageStamp.php

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaTransport.php

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaTransportFactory.php

jderusse · 2021-01-05T10:16:14Z

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaTransportFactory.php

+        $conf = new KafkaConf();
+
+        // Set a rebalance callback to log partition assignments (optional)
+        $conf->setRebalanceCb($this->createRebalanceCb($this->logger));


Does it have to be a \Closure couldn't it be a callable? $conf->setRebalanceCb([$this, 'rebalanceCb']);

Afaik it can be a callable.

I changed it :)

jderusse · 2021-01-05T10:18:47Z

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaTransportFactory.php

+        // Set a rebalance callback to log partition assignments (optional)
+        $conf->setRebalanceCb($this->createRebalanceCb($this->logger));
+
+        $brokers = $this->stripProtocol($dsn);


Dns is not only about hosts, other bdriges uses queryString to provide parameters.
ie: kafka://host:port?flushTimeout=5000

jderusse · 2021-01-05T10:20:25Z

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaTransportFactory.php

+    private function stripProtocol(string $dsn): array
+    {
+        $brokers = [];
+        foreach (explode(',', $dsn) as $currentBroker) {


sounds fragile to me. What is somebody send kafka://host?param=option1,option2

To provide multiple hosts I would be consistent with redis DSN in cache component

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaTransportFactory.php

src/Symfony/Component/Messenger/Bridge/Kafka/composer.json

src/Symfony/Component/Messenger/Bridge/Kafka/README.md

src/Symfony/Component/Messenger/Bridge/Kafka/Tests/Transport/KafkaTransportIntegrationTest.php

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaReceiver.php

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaReceiverProperties.php

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaSender.php

src/Symfony/Component/Messenger/Bridge/Kafka/composer.json

src/Symfony/Component/Messenger/Bridge/Kafka/Tests/Transport/KafkaTransportIntegrationTest.php

src/Symfony/Component/Messenger/Bridge/Kafka/Tests/Transport/KafkaTransportTest.php

Steveb-p · 2021-01-17T00:58:32Z

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaReceiver.php

+                    'key' => $message->key,
+                    'partition' => $message->partition,
+                    'offset' => $message->offset,
+                    'timestamp' => $message->timestamp,


@jderusse I second the opinion that those keys are valuable and often used, especially when cross-language communication between applications is needed. As you can see values passed to serializer come from Kafka Message object, which would be otherwise unavailable to end user.

Ping @nick-zh.

Steveb-p · 2021-01-17T01:07:54Z

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaReceiver.php

+                $this->logger->info('Kafka: Partition EOF reached. Waiting for next message ...');
+                break;
+            case \RD_KAFKA_RESP_ERR__TIMED_OUT:
+                $this->logger->debug('Kafka: Consumer timeout.');


We had a lot of discussion in phprdkafka about this. It's an internal message coming from librdkafka which has a lot of different possible meanings - some are "normal", like described above (especially if partition EOF reporting is disabled via config, which it ideally shouldn't be - so it generates EOF, not this). Others might indicate a problem.

librdkafka.h described this error as:

https://docs.confluent.io/2.0.0/clients/librdkafka/rdkafka_8h.html#a03509bab51072c72a8dcf52337e6d5cba191a3d68aab046a25af5e861a5ce394e

RD_KAFKA_RESP_ERR__TIMED_OUT Operation timed out

Steveb-p · 2021-01-17T01:10:49Z

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaReceiver.php

+                $this->logger->debug('Kafka: Consumer timeout.');
+                break;
+            case \RD_KAFKA_RESP_ERR__TRANSPORT:
+                $this->logger->debug('Kafka: Broker transport failure.');


Same as above unfortunately - usually consumer can recover from this if connection can be reestablished. In this case I'd move this up to an error level, because if this happens then something wrong is happening with the broker / communication, but as said - consumer can recover from this.

Steveb-p · 2021-01-17T01:15:34Z

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaReceiver.php

+        $message = $transportStamp->getMessage();
+
+        if ($this->properties->isCommitAsync()) {
+            $consumer->commitAsync($message);


Using commit (synchronous) usually drops the throughput of the implementation (afaik). From my experience people that use phprdkafka usually switch to async commits strictly because the difference in speed it has.

@nick-zh can you support / deny me on this?

Steveb-p · 2021-01-17T01:24:41Z

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaSender.php

+            $payload['body'],
+            $payload['key'] ?? null,
+            $payload['headers'] ?? null,
+            $payload['timestamp_ms'] ?? null


It's allowed by phprdkafka:

timestamp_ms
Timestamp that should be set for the message, if not set, the broker will set it.

In this case, it should be omitted or read from a stamp.

Steveb-p · 2021-01-17T01:27:22Z

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaTransportFactory.php

+        $conf = new KafkaConf();
+
+        // Set a rebalance callback to log partition assignments (optional)
+        $conf->setRebalanceCb($this->createRebalanceCb($this->logger));


Afaik it can be a callable.

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaSender.php

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaTransport.php

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaTransportFactory.php

src/Symfony/Component/Messenger/Bridge/Kafka/composer.json

OskarStark · 2022-12-08T22:15:52Z

You are welcome 😃

derrabus · 2023-07-22T21:23:45Z

FYI, a second PR has been opened on this topic: #51070.

KonstantinCodes · 2023-07-31T09:24:49Z

@derrabus Oh that's interesting! This PR is just waiting for a review at the moment.

chalasr · 2024-04-06T11:35:28Z

Although I think it'd have been better to contribute to this PR instead of opening a new (competing) one, the implementation in #51070 seems more inline with the main component design and what we have in other bridges. I'm also pretty sure some concerns addressed here helped making the other implementation better. All in all, I'm going to take the easy path and close this one in favor of #51070. Sorry about that, thanks a lot for your work on this.

KonstantinCodes requested a review from sroze as a code owner January 4, 2021 16:02

carsonbot added Status: Needs Review Feature labels Jan 4, 2021

KonstantinCodes changed the title ~~Kafka Transport Bridge~~ [Messenger] Kafka Transport Bridge Jan 4, 2021

KonstantinCodes force-pushed the feature/kafka-messenger branch from c5d506c to 38ae438 Compare January 4, 2021 16:10

nicolas-grekas added this to the 5.x milestone Jan 4, 2021

derrabus added the Messenger label Jan 4, 2021

derrabus reviewed Jan 4, 2021

View reviewed changes

src/Symfony/Component/Messenger/Bridge/Kafka/composer.json Outdated Show resolved Hide resolved

derrabus reviewed Jan 4, 2021

View reviewed changes

src/Symfony/Component/Messenger/Bridge/Kafka/composer.json Outdated Show resolved Hide resolved

derrabus reviewed Jan 4, 2021

View reviewed changes

src/Symfony/Component/Messenger/Bridge/Kafka/composer.json Outdated Show resolved Hide resolved

derrabus reviewed Jan 4, 2021

View reviewed changes

src/Symfony/Component/Messenger/Bridge/Kafka/composer.json Outdated Show resolved Hide resolved

derrabus reviewed Jan 4, 2021

View reviewed changes

src/Symfony/Component/Messenger/Bridge/Kafka/phpstan.neon Outdated Show resolved Hide resolved

derrabus requested changes Jan 4, 2021

View reviewed changes

carsonbot added Status: Needs Work and removed Status: Needs Review labels Jan 4, 2021

KonstantinCodes force-pushed the feature/kafka-messenger branch 3 times, most recently from 846e72b to 97ce904 Compare January 5, 2021 10:15

jderusse reviewed Jan 5, 2021

View reviewed changes

OskarStark reviewed Jan 6, 2021

View reviewed changes

KonstantinCodes force-pushed the feature/kafka-messenger branch 4 times, most recently from 9aea4b6 to f00af8a Compare January 6, 2021 11:14

babeuloula reviewed Jan 8, 2021

View reviewed changes

src/Symfony/Component/Messenger/Bridge/Kafka/composer.json Outdated Show resolved Hide resolved

Steveb-p reviewed Jan 17, 2021

View reviewed changes

OskarStark reviewed Nov 20, 2022

View reviewed changes

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaSender.php Outdated Show resolved Hide resolved

OskarStark reviewed Nov 20, 2022

View reviewed changes

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaTransport.php Outdated Show resolved Hide resolved

OskarStark reviewed Nov 20, 2022

View reviewed changes

src/Symfony/Component/Messenger/Bridge/Kafka/Transport/KafkaTransportFactory.php Outdated Show resolved Hide resolved

OskarStark reviewed Nov 20, 2022

View reviewed changes

src/Symfony/Component/Messenger/Bridge/Kafka/composer.json Outdated Show resolved Hide resolved

KonstantinCodes force-pushed the feature/kafka-messenger branch 8 times, most recently from 9a372cc to 0a89864 Compare December 17, 2022 08:52

KonstantinCodes force-pushed the feature/kafka-messenger branch 6 times, most recently from 043b54f to b2f3785 Compare February 4, 2023 04:47

Kafka Transport Bridge

3614ebd

KonstantinCodes force-pushed the feature/kafka-messenger branch from b2f3785 to 3614ebd Compare February 4, 2023 04:48

nicolas-grekas modified the milestones: 6.3, 6.4 May 23, 2023

andythorne mentioned this pull request Jul 21, 2023

[Messenger] Kafka Transport Bridge #51070

Open

nicolas-grekas modified the milestones: 6.4, 7.1 Nov 15, 2023

chalasr closed this Apr 6, 2024

Uh oh!

[Messenger] Kafka Transport Bridge #39712

[Messenger] Kafka Transport Bridge #39712

Uh oh!

Conversation

KonstantinCodes commented Jan 4, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

derrabus left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

KonstantinCodes commented Jan 4, 2021

Uh oh!

derrabus commented Jan 4, 2021

Uh oh!

KonstantinCodes commented Jan 4, 2021

Uh oh!

derrabus commented Jan 4, 2021

Uh oh!

jderusse left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

KonstantinCodes commented Jan 4, 2021 •

edited

Loading