Problems with consuming messages from the Kafka topic

[denyskanunnikov]

Good evening, I request assistance and advice in resolving the following issue. On the development host machine, a topic "se-collector.pt.event.out" has been created in the context of brokers, within which events are stored.

I am using the following RoadRunner configuration:

version: "3"

server:
  relay: pipes
  command: "/usr/bin/php -d xdebug.mode=debug -d xdebug.start_with_request=yes -d xdebug.client_host=host.docker.internal -d xdebug.client_port=\"9008\" /app/consumer.php"

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

jobs:
  consume: ["complayer-queue"]
  num_pollers: 1
  pipeline_size: 1000
  pool:
    num_workers: 1
    allocate_timeout: 60s
  pipelines:
    complayer-queue:
      driver: kafka
      config:
        consumer_options:
          topics: ["se-collector.pt.event.out"]
        group_options:
          group_id: "complayerv2.0-test-consumer"
rpc:
  enable: true
  listen: tcp://127.0.0.1:6001

Example queue consumer

use Spiral\RoadRunner\Jobs\Consumer;
use Spiral\RoadRunner\Jobs\Task\ReceivedTaskInterface;

require_once __DIR__.'/vendor/autoload.php';
$consumer = new Consumer();
echo "Consumer was created ...";
/** @var Spiral\RoadRunner\Jobs\Task\ReceivedTaskInterface $task */
while ($task = $consumer->waitTask()) {
    var_dump($task);
}
echo "Events were consumed ...";

However, this script does not consume messages from the broker.

[rustatian]

Hey @denyskanunnikov 👋🏻

Could you please try to remove the xdebug stuff from the command options? And also, what is your RR version?

Also, there is no enable: true option for the RPC plugin.

[rustatian]

Could you please also try the first 2023.3 Release Candidate? I suspect that there are some problems with the connection between the RR <-> Kafka broker. We've added a ping test before starting the plugin.
(And try to remove xdebug as well 😄 )

[denyskanunnikov]

Event was published to Kafka topic without roadrunner using another program package. I want only consuming messages with roadrunner workers

[denyskanunnikov]

Sorry, I don't understand which version should be specified, as I'm using the latest one from the
https://hub.docker.com/r/spiralscout/roadrunner/

[rustatian]

Here: https://hub.docker.com/r/spiralscout/roadrunner/tags
You need to use the 2023.3.0-rc.1 tag.

[denyskanunnikov]

Thank you, very much! I understand with were is problem with connection to broker

/rr serve -c /etc/rr.yaml
2023-09-29T18:09:04+0000        DEBUG   rpc             plugin was started      {"address": "tcp://127.0.0.1:6001", "list of the plugins with RPC methods:": ["jobs", "app", "resetter", "informer", "lock"]}
2023-09-29T18:09:04+0000        DEBUG   jobs            initializing driver     {"pipeline": "complayer-queue", "driver": "kafka"}
2023-09-29T18:09:04+0000        DEBUG   kafka           ping kafka: ok  {"driver": "kafka", "pipeline": "complayer-queue"}
2023-09-29T18:09:04+0000        DEBUG   jobs            driver ready    {"pipeline": "complayer-queue", "driver": "kafka", "start": "2023-09-29T18:09:04+0000", "elapsed": "3.339499ms"}
2023-09-29T18:09:04+0000        DEBUG   kafka           pipeline was started    {"driver": "kafka", "pipeline": "complayer-queue", "start": "2023-09-29T18:09:04+0000", "elapsed": "2.538µs"}
2023-09-29T18:09:04+0000        INFO    server          Xdebug: [Step Debug] Could not connect to debugging client. Tried: host.docker.internal:9008 (through xdebug.client_host/xdebug.client_port).

2023-09-29T18:09:04+0000        INFO    server          Consumer was created ...
2023-09-29T18:09:04+0000        DEBUG   server          worker is allocated     {"pid": 26, "internal_event_name": "EventWorkerConstruct"}
2023-09-29T18:09:04+0000        DEBUG   jobs            exited from jobs processor
2023-09-29T18:09:04+0000        DEBUG   jobs            exited from jobs processor
2023-09-29T18:09:04+0000        DEBUG   jobs            exited from jobs processor
2023-09-29T18:09:04+0000        DEBUG   jobs            exited from jobs processor
2023-09-29T18:09:04+0000        DEBUG   jobs            exited from jobs processor
2023-09-29T18:09:04+0000        DEBUG   jobs            exited from jobs processor
2023-09-29T18:09:04+0000        DEBUG   jobs            job processing was started      {"ID": "500000004", "start": "2023-09-29T18:09:04+0000", "elapsed": "77.17µs"}
[INFO] RoadRunner server started; version: 2023.3.0-rc.1, buildtime: 2023-09-14T17:17:56+0000
[INFO] sdnotify: not notified
2023-09-29T18:09:04+0000        DEBUG   jobs            exited from jobs processor
2023-09-29T18:09:04+0000        DEBUG   jobs            exited from jobs processor
2023-09-29T18:09:04+0000        DEBUG   jobs            exited from jobs processor
2023-09-29T18:09:04+0000        DEBUG   jobs            exited from jobs processor
2023-09-29T18:09:04+0000        INFO    server          /app/consumer.php:10:
class Spiral\RoadRunner\Jobs\Task\ReceivedTask#10 (8) {
  protected string $name =>
  string(13) "deduced_by_rr"
  protected array $payload =>
  array(4) {
    'EventID' =>
    string(16) "tt_6516d2411ff7e"
    'Type' =>
    string(18) "SOURCE_ESTATE_DIFF"
    'DataModel' =>
    string(18) "SOURCE_ESTATE_DIFF"
    'Data' =>
    string(7924) "eyJiZWZvcmUiOm51bGwsImFmdGVyIjp7ImluaXRpYWwiOnsiYmFzZSI6eyJzZUlEIjo1MDAwMDAwMDQsImFwcElEIjoxLCJzb3VyY2VJRCI6MTB9LCJ0aW1lIjoiMjAyMy0wOS0yOVQxMzozMzo1M1oiLCJ1aWQiOiJ1aWRfMjY3MCJ9LCJhZ2VuY3lEYXRhIjp7InRpbWUiOiIyMDIzLTA5LTI5VDEzOjMzOjUzWiIsImNvbnRhY3RzIjoiZXhhbXBsZUB5aG9vLmNvbSIsImNvbXBhbnkiOiJNZW5vcmNhIFNvdGhlYnkncyBJbnRlcm5hdGlvbmFsIFJlYWx0eVxuICAgICAgICAgICAgICAgIElzYWJlbCBNbyIsInBob25lIjoiMDYgNzQgNjMgNTAgMTEiLCJpc1ByaXZhdGUiOnRydWV9LCJjb21mb3J0U3BlY3MiOnsidGltZSI6IjIwMjMtMDktMjlUMTM6MzM6NTNaIiwicGFya2luZyI6"...
  }
  protected array $headers =>
  array(2) {
    'Traceparent' =>
    array(1) {
      [0] =>
      string(55) "00-79338d58541b36bd055a9e99cd4cd8c1-38a56d97371e06e2-01"
    }
    'Uber-Trace-Id' =>
    array(1) {
      [0] =>
      string(53) "79338d58541b36bd055a9e99cd4cd8c1:38a56d97371e06e2:0:1"
    }
  }
  protected string $id =>
  string(9) "500000004"
  protected string $queue =>
  string(13) "deduced_by_rr"
  private Spiral\RoadRunner\WorkerInterface $worker =>
  class Spiral\RoadRunner\Worker#5 (2) {
    private Spiral\Goridge\RelayInterface $relay =>
    class Spiral\Goridge\StreamRelay#6 (2) {
      private $in =>
      resource(1) of type (stream)
      private $out =>
      resource(2) of type (stream)
    }
    private Psr\Log\LoggerInterface $logger =>
    class Spiral\RoadRunner\Logger#7 (0) {
    }
  }
  private ?int $completed =>
  NULL
  private int $delay =>
  int(0)
}

[rustatian]

@denyskanunnikov My pleasure ❤️