How to correctly configure Roadrunner to consume messages from a Kafka topic?

[denyskanunnikov]

Good evening, I need your assistance or consultation in resolving the following issue. In the Kafka broker a topic named "se-collector.pt.event.out" has been created. How can I correctly configure the consume/pipelines in rr.yaml to consume and process messages from the specified topic name "se-collector.pt.event.out"?
I try run Roadrunner server with sample configuration file

version: "3"
server:
  relay: pipes
  command: /usr/bin/php /app/consumer.php

kafka:
  brokers: ["dev-producer_api-1:8001"]

jobs:
  consume: ["se-collector.pt.event.out"]
  num_pollers: 1
  pipeline_size: 1000
  pool:
    num_workers: 1
    max_jobs: 0
    allocate_timeout: 60s
    destroy_timeout: 60s
  pipelines:
    "se-collector.pt.event.out":
      driver: kafka
      config:
        priority: 10
        prefetch: 10

rpc:
  enable: true
  listen: tcp://127.0.0.1:6001

but got errors like

root@e9af06c22ec5:/app# /usr/local/bin/rr serve -c /etc/rr.yaml
2023-09-18T15:54:15+0000        DEBUG   rpc             plugin was started      {"address": "tcp://127.0.0.1:6001", "list of the plugins with RPC methods:": ["informer", "metrics", "lock", "app", "jobs", "resetter"]}
2023-09-18T15:54:15+0000        WARN    jobs            can't find driver name for the pipeline, please, check that the 'driver' keyword for the pipelines specified correctly, JOBS plugin will try to run the next pipeline

[rustatian]

Hey @denyskanunnikov 👋🏻
You should specify consumers (consume groups)/producers in the configuration. You may read more here: https://roadrunner.dev/docs/queues-kafka/current/en

[rustatian]

Pipeline name is not the same as the topic or queue names. These are RR internal names. So, in your case, this name should be here: https://github.com/roadrunner-server/roadrunner/blob/master/.rr.yaml#L1718

[denyskanunnikov]

Thank you, please also advise, in the consume/pipelines blocks, should I simply specify a unique key value without any dependency on the broker's topics?

[rustatian]

Yeah 😃. Just keep in mind, you may give them (pipelines) any name you want. This is only your name to recognize what's happening inside these pipelines.
Consume array used to start consuming messages after RR starts. If you want to use pipeline to only push the messages, you don't need to put the pipeline name into cosume array.
To test everything, you may use in-memory driver. It's the same as Kafka, but located in memory. From your app perspective, you'll receive the same messages.

[konigbach]

Hello! I am a little bit lost regarding the consumer part: I understand that you defined the topics to be consumed in the config file but, which php class is going to handle it? Let's say I have a topic 'users' with different messages ('user-created', 'user-deleted'...). How do I set 'this topic is going to be handled by this class'? I know how to work with Spiral jobs (based on names) but consuming external messages is a mystery. Thank you! :)

[rustatian]

Hey @konigbach 👋
Please, create a new discussion or topic in our discord forum for your case 😃