Consuming with Simple Client and Kafka

[cabralc]

I'm having some trouble consuming messages with the SimpleClient and the rdkafka transport.

Here is my config file:

return [
        'transport' => [
            'default' => 'rdkafka',
            'rdkafka' => [
                'global' => [
                    'group.id'          => 'app',
                    'bootstrap.servers' => env('KAFKA_BROKERS'),
                    'security.protocol' => 'sasl_ssl',
                    'sasl.mechanism'    => 'SCRAM-SHA-256',
                    'sasl.username'     => env('KAFFKA_USERNAME'),
                    'sasl.password'     => env('KAFKA_PASSWORD'),
                ],
                'topic'  => [
                ],
            ],
        ],
];

And here is the code I'm trying:

   /** @var \Enqueue\SimpleClient\SimpleClient $client */
    $client = new SimpleClient(config('enqueue.client'));
    $client->setupBroker();

    $message = new Message();
    $message->setScope(Message::SCOPE_MESSAGE_BUS);
    $message->setBody('message');
    $client->sendEvent('foo', $message);

    $client->bind('foo', KafkaProcessor::class, function(PsrMessage $psrMessage) {

        return PsrProcessor::ACK;
    });

    $client->consume()

Using the default config when producing it sends the message successfully to the "enqueue.default" topic. Then when consuming it creates a new topic "enqueue.app.default" and it never consumes the messages from "enqueue.default".

Am I missing something with the configuration? Or I am not understanding the difference between the app and message bus scope? Any insight would be extremely helpful. Thanks.

[makasim]

try to make a router topic and route queue to be the same

return [
   // your part
  'client' => [
    'prefix' => '',
    'app_name' => '',
  ]
];

There will be a proper fixed in upcoming version.

[cabralc]

Thank you that works.

I have another question. When producing the message it seems to choose the partition at random. Is there a way to specify the partition you are producing and consuming from?

Also when creating the topic it creates the topic with a certain number of partitions. I haven't figured out if this is a configuration within kafka or from the code that is creating the topic.

Again any insight into this would be a huge help. Thanks again.

[makasim]

Partition could be set on a message basis

https://github.com/php-enqueue/enqueue-dev/blob/master/pkg/rdkafka/RdKafkaProducer.php#L39

On a consumer side it could be set too

https://github.com/php-enqueue/enqueue-dev/blob/master/pkg/rdkafka/RdKafkaConsumer.php#L94
https://github.com/php-enqueue/enqueue-dev/blob/master/pkg/rdkafka/RdKafkaConsumer.php#L69