Don't read tasks from Roadrunner's consumer, need help!

[sayankanatov]

I am new in Roadrunner (ghcr.io/roadrunner-server/roadrunner:2024.3.5). Wrote simple consumer:

$consumer = new \Spiral\RoadRunner\Jobs\Consumer();

while ($task = $consumer->waitTask()) {
    try {
        if (!$task instanceof \Spiral\RoadRunner\Jobs\Task\ReceivedTaskInterface) {
            $logger->critical("🔥 FATAL ERROR");
        }

        $payload = json_decode($task->getPayload(), true);

        (new MyWorker($logger))->handle($payload);

        $task->complete();

    } catch (\Exception $e) {
        $task->fail($e->getMessage());
        $logger->critical("🔥 FATAL ERROR: " . $e->getMessage());
    }
}

My .rr.yaml:

version: "3"

rpc:
    listen: "tcp://0.0.0.0:6001"
    log_level: debug
    enable: true

server:
    command: "php worker-rr.php"
    relay: "pipes"
    env:
        APP_RUNTIME: Runtime\RoadRunnerSymfonyNyholm\Runtime

plugins:
    jobs: true
    kafka: true

kafka:
    brokers: ["10.254.3.89:9092", "10.254.3.88:9092", "10.254.3.90:9092"]
    ping:
        timeout: "10s"
    local-memory:
        driver: memory
        config: { }

jobs:
    consume:
        - kafka_pipeline
        - in_memory
    num_pollers: 1
    pipeline_size: 1000
    pool:
        num_workers: 1
        max_jobs: 0
        allocate_timeout: 60s
        destroy_timeout: 60s
    pipelines:
        in_memory:
            # In memory driver specific section
            #
            # Default: none
            driver: memory
            config: { }
        kafka_pipeline:
            driver: kafka
            config:
                priority: 10
                auto_offset_reset: "earliest"
                auto_create_topics_enable: false
            producer_options:
                disable_idempotent: false
                required_acks: AllISRAck
                max_message_bytes: 1000012
                request_timeout: 10s
                delivery_timeout: 100s
                transaction_timeout: 100
                compression_codec: gzip
            consumer_options:
                topics: ["my_topic"]
                consume_partitions:
                    my_topic:
                        0:
                            type: AtStart
                            value: 1
                        1:
                            type: AtStart
                            value: 1
                        2:
                            type: AtStart
                            value: 1
                        3:
                            type: AtStart
                            value: 1
            group_options:
                group_id: my.group

logs:
  mode: development
  level: debug

After starting a connection, I received these logs:

2025-03-16 16:23:52 2025-03-16T10:23:52+0000    DEBUG   rpc             plugin was started      {"address": "tcp://0.0.0.0:6001", "list of the plugins with RPC methods:": ["app", "jobs", "resetter", "lock", "informer"]}
2025-03-16 16:23:52 2025-03-16T10:23:52+0000    DEBUG   jobs            initializing driver     {"pipeline": "in_memory", "driver": "memory"}
2025-03-16 16:23:52 2025-03-16T10:23:52+0000    DEBUG   jobs            initializing driver     {"pipeline": "kafka_pipeline", "driver": "kafka"}
2025-03-16 16:23:52 2025-03-16T10:23:52+0000    DEBUG   jobs            driver ready    {"pipeline": "in_memory", "driver": "memory", "start": "2025-03-16T10:23:52+0000", "elapsed": 1}
2025-03-16 16:23:52 2025-03-16T10:23:52+0000    DEBUG   memory          pipeline was started    {"driver": "memory", "pipeline": "in_memory", "start": "2025-03-16 10:23:52.76815788 +0000 UTC", "elapsed": "14.405µs"}
2025-03-16 16:23:52 2025-03-16T10:23:52+0000    DEBUG   kafka.kgo       opening connection to broker    {"kgo_driver": "addr", "kgo_driver": "10.254.3.89:9092", "kgo_driver": "broker", "kgo_driver": "seed_0"}
2025-03-16 16:23:52 2025-03-16T10:23:52+0000    DEBUG   kafka.kgo       connection opened to broker     {"kgo_driver": "addr", "kgo_driver": "10.254.3.89:9092", "kgo_driver": "broker", "kgo_driver": "seed_0"}
2025-03-16 16:23:52 2025-03-16T10:23:52+0000    DEBUG   kafka.kgo       issuing api versions request    {"kgo_driver": "broker", "kgo_driver": "seed_0", "kgo_driver": "version", "kgo_driver": 3}
2025-03-16 16:23:52 2025-03-16T10:23:52+0000    DEBUG   kafka.kgo       wrote ApiVersions v3    {"kgo_driver": "broker", "kgo_driver": "seed_0", "kgo_driver": "bytes_written", "kgo_driver": 31, "kgo_driver": "write_wait", "kgo_driver": "39.74µs", "kgo_driver": "time_to_write", "kgo_driver": "54.029µs", "kgo_driver": "err", "kgo_driver": null}
2025-03-16 16:23:52 2025-03-16T10:23:52+0000    DEBUG   kafka.kgo       read ApiVersions v3     {"kgo_driver": "broker", "kgo_driver": "seed_0", "kgo_driver": "bytes_read", "kgo_driver": 418, "kgo_driver": "read_wait", "kgo_driver": "111.589µs", "kgo_driver": "time_to_read", "kgo_driver": "9.362718ms", "kgo_driver": "err", "kgo_driver": null}
2025-03-16 16:23:52 2025-03-16T10:23:52+0000    DEBUG   kafka.kgo       connection initialized successfully     {"kgo_driver": "addr", "kgo_driver": "10.254.3.89:9092", "kgo_driver": "broker", "kgo_driver": "seed_0"}
2025-03-16 16:23:52 2025-03-16T10:23:52+0000    DEBUG   kafka.kgo       wrote ApiVersions v3    {"kgo_driver": "broker", "kgo_driver": "seed_0", "kgo_driver": "bytes_written", "kgo_driver": 31, "kgo_driver": "write_wait", "kgo_driver": "21.810228ms", "kgo_driver": "time_to_write", "kgo_driver": "34.91µs", "kgo_driver": "err", "kgo_driver": null}
2025-03-16 16:23:52 2025-03-16T10:23:52+0000    DEBUG   kafka.kgo       read ApiVersions v3     {"kgo_driver": "broker", "kgo_driver": "seed_0", "kgo_driver": "bytes_read", "kgo_driver": 418, "kgo_driver": "read_wait", "kgo_driver": "116.914µs", "kgo_driver": "time_to_read", "kgo_driver": "8.295543ms", "kgo_driver": "err", "kgo_driver": null}
2025-03-16 16:23:52 2025-03-16T10:23:52+0000    DEBUG   kafka           ping kafka: ok  {"driver": "kafka", "pipeline": "kafka_pipeline"}
2025-03-16 16:23:52 2025-03-16T10:23:52+0000    DEBUG   jobs            driver ready    {"pipeline": "kafka_pipeline", "driver": "kafka", "start": "2025-03-16T10:23:52+0000", "elapsed": 31}
2025-03-16 16:23:52 2025-03-16T10:23:52+0000    DEBUG   kafka           pipeline was started    {"driver": "kafka", "pipeline": "kafka_pipeline", "start": "2025-03-16T10:23:52+0000", "elapsed": 0}
2025-03-16 16:23:53 2025-03-16T10:23:53+0000    DEBUG   server          worker is allocated     {"pid": 17, "max_execs": 0, "internal_event_name": "EventWorkerConstruct"}
2025-03-16 16:23:53 [INFO] RoadRunner server started; version: 2024.3.5, buildtime: 2025-02-27T17:24:27+0000
2025-03-16 16:23:53 [INFO] sdnotify: not notified
2025-03-16 16:24:32 2025-03-16T10:24:32+0000    DEBUG   kafka.kgo       reaped connections      {"kgo_driver": "time_since_last_reap", "kgo_driver": "19.999486276s", "kgo_driver": "reap_dur", "kgo_driver": "17.317633ms", "kgo_driver": "num_reaped", "kgo_driver": 1}
2025-03-16 16:28:52 2025-03-16T10:28:52+0000    DEBUG   kafka.kgo       opening connection to broker    {"kgo_driver": "addr", "kgo_driver": "10.254.3.89:9092", "kgo_driver": "broker", "kgo_driver": "seed_0"}
2025-03-16 16:28:52 2025-03-16T10:28:52+0000    DEBUG   kafka.kgo       connection opened to broker     {"kgo_driver": "addr", "kgo_driver": "10.254.3.89:9092", "kgo_driver": "broker", "kgo_driver": "seed_0"}
2025-03-16 16:28:52 2025-03-16T10:28:52+0000    DEBUG   kafka.kgo       connection initialized successfully     {"kgo_driver": "addr", "kgo_driver": "10.254.3.89:9092", "kgo_driver": "broker", "kgo_driver": "seed_0"}
2025-03-16 16:28:52 2025-03-16T10:28:52+0000    DEBUG   kafka.kgo       wrote Metadata v11      {"kgo_driver": "broker", "kgo_driver": "seed_0", "kgo_driver": "bytes_written", "kgo_driver": 22, "kgo_driver": "write_wait", "kgo_driver": "23.938852ms", "kgo_driver": "time_to_write", "kgo_driver": "98.515µs", "kgo_driver": "err", "kgo_driver": null}
2025-03-16 16:28:52 2025-03-16T10:28:52+0000    DEBUG   kafka.kgo       read Metadata v11       {"kgo_driver": "broker", "kgo_driver": "seed_0", "kgo_driver": "bytes_read", "kgo_driver": 109, "kgo_driver": "read_wait", "kgo_driver": "303.243µs", "kgo_driver": "time_to_read", "kgo_driver": "33.194956ms", "kgo_driver": "err", "kgo_driver": null}
2025-03-16 16:29:32 2025-03-16T10:29:32+0000    DEBUG   kafka.kgo       reaped connections      {"kgo_driver": "time_since_last_reap", "kgo_driver": "20.000016054s", "kgo_driver": "reap_dur", "kgo_driver": "815.722µs", "kgo_driver": "num_reaped", "kgo_driver": 1}

Terminal's display:

c88747e07a8b:/app# rr workers
Workers of [jobs]:
+---------+-----------+---------+---------+---------+--------------------+
|   PID   |  STATUS   |  EXECS  | MEMORY  |  CPU%   |      CREATED       |
+---------+-----------+---------+---------+---------+--------------------+
|      17 | ready     |       0 | 20 MB   |    0.06 | 1 minute ago       |
+---------+-----------+---------+---------+---------+--------------------+
Jobs of [jobs]:
+--------+----------------+--------+-----------+--------+---------+----------+
| STATUS |    PIPELINE    | DRIVER |   QUEUE   | ACTIVE | DELAYED | RESERVED |
+--------+----------------+--------+-----------+--------+---------+----------+
| READY  | in_memory      | memory | in_memory | 0      | 0       | 0        |
| READY  | kafka_pipeline | kafka  | NO TOPICS | 0      | 0       | 0        |
+--------+----------------+--------+-----------+--------+---------+----------+
c88747e07a8b:/app# ```

What am I doing wrong? and why I don't see a topics in kafka_pipeline?

[rustatian]

Hey @sayankanatov 👋🏻
Where did you get this configuration? You have a 50/50 mix of real and random (non-existent) options in the configuration.

[sayankanatov]

Main resource: https://github.com/roadrunner-server/roadrunner/blob/master/.rr.yaml. What I need to add/delete?

[rustatian]

Please, check the main source and double-check your configuration.

[sayankanatov]

blocks producer_options and consumer_options should be under config block. This is solution!