[Messenger] Kafka Transport Bridge #39712

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

Q	A
Branch?	5.4
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.

jderusse · 2021-01-05T10:24:07Z

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

@@ -0,0 +1,10 @@
+<?php
+
+declare(strict_types=1);


We don't use strict type in symfony

I confirm, this should be removed

jderusse · 2021-01-05T10:24:07Z

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:24:07Z

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:24:07Z

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:24:07Z

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-17T01:29:04Z

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:29:04Z

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:29:04Z

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:29:04Z

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:29:04Z

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:29:04Z

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.

jderusse

I do think the stamp is needed to convoy extra informations like key, partition, timestamp_ms, ...

jderusse · 2021-10-29T07:31:09Z

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

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


Well, $payload is the result of $this->serializer->encode($envelope).
The serializers provided by symfony only returns body and headers.

The fields key and timestamp_ms used here does not exists.
If you want extra data, you should use a custom Stamp

jderusse · 2021-10-29T07:31:09Z

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

+        $this->conf->setRebalanceCb($this->createRebalanceCb($this->logger));
+    }
+
+    public function get(): iterable


The contract of this method is defined by the Interface

jderusse · 2021-10-29T07:31:09Z

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

+                    'body' => $message->payload,
+                    'headers' => $message->headers,
+                    'key' => $message->key,
+                    'topic_name' => $message->topic_name,
+                    'partition' => $message->partition,
+                    'offset' => $message->offset,
+                    'timestamp' => $message->timestamp,


most of serializers will only take care of headers and body

fabpot · 2022-03-26T14:29:15Z

Is it something we want to include for 6.1?

KonstantinCodes · 2022-03-29T08:21:43Z

@fabpot I'll rebase the branch and fix the tests :)

torbenbr · 2022-08-17T06:08:24Z

@KonstantinCodes What's the problem that you don't continue here? Do you need help or something? I would be very happy if continued here <3

KonstantinCodes · 2022-08-17T06:15:22Z

@torbenbr hey, thank you for asking! Actually, the build system was puzzling me. Maybe we can take a look at that together?

torbenbr · 2022-08-17T06:45:33Z

@KonstantinCodes I'll look at it tonight (Timezone UTC+2)

KonstantinCodes requested a review from sroze as a code owner Jan 4, 2021

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 Jan 4, 2021

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 changes

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

derrabus reviewed Jan 4, 2021

View changes

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

derrabus reviewed Jan 4, 2021

View changes

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

derrabus reviewed Jan 4, 2021

View changes

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

derrabus reviewed Jan 4, 2021

View changes

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

derrabus requested changes Jan 4, 2021

View 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 Jan 5, 2021

jderusse reviewed Jan 5, 2021

View changes

OskarStark reviewed Jan 6, 2021

View changes

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

babeuloula reviewed Jan 8, 2021

View changes

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

Steveb-p reviewed Jan 17, 2021

View changes

jderusse reviewed Oct 29, 2021

View changes

KonstantinCodes added 8 commits Oct 29, 2021

Kafka Transport Bridge

6e735b7

Psalm: New Ubuntu

d608c6f

Psalm: Remove Confluent Repo

af592c1

Allow 6.0

f594522

Use "kafka://" as protocol

2445607

codestyle + handle stamp being null

ff0d59b

flush retries loop log only at failure

9b049f7

codestyle

d04c435

KonstantinCodes force-pushed the feature/kafka-messenger branch from 3f019ca to d04c435 Compare Oct 29, 2021

KonstantinCodes added 8 commits Oct 29, 2021

Add license to test files

3448b1a

adjust timeout in test

feb2457

psalm

5f8b137

psalm

ab58338

psalm

51f8486

psalm

ae4785f

psalm

d581dd8

psalm

cd59f23

fabpot modified the milestones: 5.4, 6.1 Nov 16, 2021

javiereguiluz mentioned this pull request Dec 8, 2021

[Messenger] Kafka Transport symfony/symfony-docs#15884

Open

gayansanjeewa mentioned this pull request Mar 22, 2022

[Messenger][RFC] Kafka transport #35549

Open

fabpot modified the milestones: 6.1, 6.2 May 20, 2022

rejinka mentioned this pull request Jul 19, 2022

Simple KafkaConsumer brefphp/symfony-messenger#62

Closed

[Messenger] Kafka Transport Bridge #39712

[Messenger] Kafka Transport Bridge #39712

KonstantinCodes commented Jan 4, 2021 •

edited

derrabus left a comment

KonstantinCodes commented Jan 4, 2021

derrabus commented Jan 4, 2021

KonstantinCodes commented Jan 4, 2021

derrabus commented Jan 4, 2021

jderusse left a comment

jderusse Jan 5, 2021

nicolas-grekas Oct 28, 2021

jderusse Jan 5, 2021

KonstantinCodes May 3, 2021

jderusse Jan 5, 2021

Steveb-p Jan 17, 2021

KonstantinCodes Jan 18, 2021

jderusse Jan 5, 2021

jderusse Jan 5, 2021

Steveb-p Jan 17, 2021

Steveb-p Jan 17, 2021

Steveb-p Jan 17, 2021

Steveb-p Jan 17, 2021

Steveb-p Jan 17, 2021

Steveb-p Jan 17, 2021

jderusse left a comment

jderusse Oct 29, 2021

jderusse Oct 29, 2021

jderusse Oct 29, 2021

fabpot commented Mar 26, 2022

KonstantinCodes commented Mar 29, 2022

torbenbr commented Aug 17, 2022

KonstantinCodes commented Aug 17, 2022

torbenbr commented Aug 17, 2022

		@@ -0,0 +1,10 @@
		<?php

		declare(strict_types=1);

[Messenger] Kafka Transport Bridge #39712

Are you sure you want to change the base?

[Messenger] Kafka Transport Bridge #39712

Conversation

KonstantinCodes commented Jan 4, 2021 • edited

derrabus left a comment

KonstantinCodes commented Jan 4, 2021

derrabus commented Jan 4, 2021

KonstantinCodes commented Jan 4, 2021

derrabus commented Jan 4, 2021

jderusse left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jderusse left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

fabpot commented Mar 26, 2022

KonstantinCodes commented Mar 29, 2022

torbenbr commented Aug 17, 2022

KonstantinCodes commented Aug 17, 2022

torbenbr commented Aug 17, 2022

KonstantinCodes commented Jan 4, 2021 •

edited