How to configure kafka job queue #1820

suityou01 · 2023-12-21T10:59:29Z

suityou01
Dec 21, 2023

I am having problems just getting the basic config set up.

version: '3'

rpc:
  listen: tcp://127.0.0.1:6001

server:
  command: "php consumer.php"

#http:
#  address: "0.0.0.0:8080"
kafka:
  addrs: 127.0.0.1:9092

jobs:
  num_pollers: 1
  pipeline_size: 100
  pool:
    num_workers: 1
    max_jobs: 0
    allocate_timeout: 60s
    destroy_timeout: 60s
  consume: ["kafka"]
  pipelines:
    kafka:
      driver: kafka
      config:
        priority: 1
        topic: feeds.feeds.unit
        brokers: [ broker:9092 ]
        consumer_options:
          topics: [ "feeds", "feeds.feeds.unit", "^[a-zA-Z0-9._-]+$" ]
          consume_regexp: true
logs:
  level: error

Error "Did not find expected key by parsing a block mapping"

NB The example has a key called topic at the jobs.pipelines.queue-name.config level which is confusing
e.g.

# Kafka jobs driver
#
# This option is required to use Kafka driver. Addrs can contain any number of addresses separated by comma (127.0.0.1:9092,127.0.0.1:9093,...)
kafka:
  addrs: 127.0.0.1:9092

jobs:
  num_pollers: 10
  pipeline_size: 100000
  pool:
    num_workers: 10
    max_jobs: 0
    allocate_timeout: 60s
    destroy_timeout: 60s

  pipelines:
    test-local-6:
      # Driver name
      #
      # This option is required
      driver: kafka

      # Driver's configuration
      #
      # Should not be empty
      config:

        # Pipeline priority
        #
        # If the job has priority set to 0, it will inherit the pipeline's priority. Default: 10.
        priority: 1

        # Topic name: https://kafka.apache.org/intro#intro_concepts_and_terms
        #
        # This option is required and should not be empty.
        topic: test-1

So we have to specify a topic here (single topic, which one??)
And for the consumer we have to specify a topics key

How does this work?

rustatian · 2023-12-21T11:01:37Z

rustatian
Dec 21, 2023
Maintainer

Hey @suityou01 👋
Ok, looks like you're using some outdated docs. There are no topic, broker and other options for the kafka driver.

You may find the correct options here: https://roadrunner.dev/docs/queues-kafka/current/en

0 replies

suityou01 · 2023-12-21T11:06:57Z

suityou01
Dec 21, 2023
Author

This makes sense.

I have the worker package installed and yet I get this error

Fatal error: Uncaught Error: Class "Spiral\RoadRunner\Environment" not found in /app/consumer.php:10

{
    "require": {
        "spiral/roadrunner-cli": "^2.6",
        "spiral/roadrunner-http": "^3.3",
        "nyholm/psr7": "^1.8",
        "spiral/roadrunner-worker": "^3.3",
        "spiral/roadrunner-jobs": "^4.3",
        "spiral/roadrunner": "^2023.3"
    }
}

<?php

use Spiral\RoadRunner\Environment;
use Spiral\RoadRunner\Environment\Mode;

// 1. Using global env variable
$isJobsMode = $_SERVER['RR_MODE'] === 'jobs';

// 2. Using RoadRunner's API
$env = Environment::fromGlobals();

$isJobsMode = $env->getMode() === Mode::MODE_JOBS;

$consumer = new Consumer();

/** @var Spiral\RoadRunner\Jobs\Task\ReceivedTaskInterface $task */
while ($task = $consumer->waitTask()) {
    var_dump($task);
}

Any ideas?

2 replies

rustatian Dec 21, 2023
Maintainer

Try to use composer -u to update deps. But this class belongs to the spiral/roadrunner-worker package.

msmakouz Dec 21, 2023

@suityou01 Hi, try to add the Composer autoloader in your file:

require __DIR__ . '/vendor/autoload.php';

suityou01 · 2023-12-21T11:43:25Z

suityou01
Dec 21, 2023
Author

Yep problem exists between chair and monitor. :-)

working code looks like

<?php

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

include "vendor/autoload.php";

// 1. Using global env variable
$isJobsMode = $_SERVER['RR_MODE'] === 'jobs';

// 2. Using RoadRunner's API
$env = Environment::fromGlobals();

$isJobsMode = $env->getMode() === Mode::MODE_JOBS;

$consumer = new Consumer();

/** @var Spiral\RoadRunner\Jobs\Task\ReceivedTaskInterface $task */
while ($task = $consumer->waitTask()) {
    var_dump($task);
}

0 replies

suityou01 · 2023-12-21T11:45:52Z

suityou01
Dec 21, 2023
Author

Not getting any output when the messages are flowing through kafka though...

version: '3'

rpc:
  listen: tcp://127.0.0.1:6001

server:
  command: "php consumer.php"

#http:
#  address: "0.0.0.0:8080"
kafka:
  brokers: ["broker:9092"]
  
jobs:
  num_pollers: 1
  pipeline_size: 100
  pool:
    num_workers: 1
    max_jobs: 0
    allocate_timeout: 60s
    destroy_timeout: 60s
  pipelines:
    kafka:
      driver: kafka
      config:
        priority: 1
        brokers: [ broker:9092 ]
        consumer_options:
          topics: [ "feeds", "feeds.feeds.unit", "^[a-zA-Z0-9._-]+$" ]
          consume_regexp: true
logs:
  level: error

1 reply

rustatian Dec 21, 2023
Maintainer

You need debug log level to see your var_dump. Also, you have to complete the task by using $task->complete(); 😃

suityou01 · 2023-12-21T12:51:06Z

suityou01
Dec 21, 2023
Author

So here is what I get in my container log

2023-12-21T12:38:20+0000	DEBUG	rpc         	plugin was started	{"address": "tcp://127.0.0.1:6001", "list of the plugins with RPC methods:": ["lock", "resetter", "informer", "jobs", "app"]}
2023-12-21T12:38:20+0000	DEBUG	jobs        	initializing driver	{"pipeline": "kafka", "driver": "kafka"}
2023-12-21T12:38:20+0000	DEBUG	kafka       	ping kafka: ok	{"driver": "kafka", "pipeline": "kafka"}
2023-12-21T12:38:20+0000	DEBUG	jobs        	driver ready	{"pipeline": "kafka", "driver": "kafka", "start": "2023-12-21T12:38:20+0000", "elapsed": "3.843019ms"}
2023-12-21T12:38:20+0000	DEBUG	server      	worker is allocated	{"pid": 16, "internal_event_name": "EventWorkerConstruct"}
2023-12-21T12:38:20+0000	DEBUG	jobs        	exited from jobs pipeline processor
2023-12-21T12:38:20+0000	DEBUG	jobs        	exited from jobs pipeline processor
2023-12-21T12:38:20+0000	DEBUG	jobs        	exited from jobs pipeline processor
2023-12-21T12:38:20+0000	DEBUG	jobs        	exited from jobs pipeline processor
2023-12-21T12:38:20+0000	DEBUG	jobs        	exited from jobs pipeline processor
2023-12-21T12:38:20+0000	DEBUG	jobs        	exited from jobs pipeline processor
2023-12-21T12:38:20+0000	DEBUG	jobs        	exited from jobs pipeline processor
2023-12-21T12:38:20+0000	DEBUG	jobs        	exited from jobs pipeline processor
2023-12-21T12:38:20+0000	DEBUG	jobs        	exited from jobs pipeline processor
2023-12-21T12:38:20+0000	DEBUG	jobs        	exited from jobs pipeline processor
[INFO] RoadRunner server started; version: 2023.3.8, buildtime: 2023-12-14T16:05:26+0000
[INFO] sdnotify: not notified

Loop now looks like

/** @var Spiral\RoadRunner\Jobs\Task\ReceivedTaskInterface $task */
while ($task = $consumer->waitTask()) {
    var_dump($task);
    $task->complete();
}

Config now looks like

version: '3'

rpc:
  listen: tcp://127.0.0.1:6001

server:
  command: "php consumer.php"

#http:
#  address: "0.0.0.0:8080"
kafka:
  brokers: ["broker:9092"]
  
jobs:
  num_pollers: 1
  pipeline_size: 100
  pool:
    num_workers: 1
    max_jobs: 0
    allocate_timeout: 60s
    destroy_timeout: 60s
  pipelines:
    kafka:
      driver: kafka
      config:
        priority: 1
        consumer_options:
          topics: [ "feeds", "feeds.feeds.unit", "^[a-zA-Z0-9._-]+$" ]
          consume_regexp: true
logs:
  level: debug

Maybe something is coming through but I cannot see the output from var_dump if it is.
Or maybe nothing is coming through and those debug log entries are about something else. How do I tell?

0 replies

suityou01 · 2023-12-21T12:59:32Z

suityou01
Dec 21, 2023
Author

I think those debug messages are on startup. So no messages are coming through it seems.

0 replies

suityou01 · 2023-12-21T13:16:11Z

suityou01
Dec 21, 2023
Author

Kafka does not report it as a consumer so it looks like the handshake is failing. How do I get more output from this black box?

5 replies

rustatian Dec 21, 2023
Maintainer

If you see the ping kafka: ok - it means, that RR connected to Kafka. You may try to use our kafka docker-compose file to check if you set up everything correctly.
Also, make sure you're connected to the correct instance. Check if you need to set group_options as well. This is the preferred way (according to the Kafka docs) to work with the consumers.

rustatian Dec 21, 2023
Maintainer

Make sure that you have selected topics on the server. Otherwise, you may use auto_create_topics_enable: true option.

rustatian Dec 21, 2023
Maintainer

@suityou01 Oh, and do not forget to turn the consumer on: link. In your case, there should be your pipeline name: kafka. BTW, you may choose any pipeline name you want, it is not necessary to be kafka. It might be some domain-specific name, like, kafka_feeds, etc.

suityou01 Dec 21, 2023
Author

In that last link about turning on the consumer I noted it said if you intend to start the consumption manually leave this blank. Is there any documentation on how to start the consumers manually? I don't want to do this in production but I thought it may help debug this problem?

rustatian Dec 21, 2023
Maintainer

Sure: link. Queue, which is not in an active state, treated as paused. So, you need to call resume method and specify the pipeline name.

suityou01 · 2023-12-21T15:25:34Z

suityou01
Dec 21, 2023
Author

OK here is the latest. It is not collecting any messages from kafka.

version: '3'

rpc:
  listen: tcp://127.0.0.1:6001

server:
  command: "php consumer.php"

#http:
#  address: "0.0.0.0:8080"
kafka:
  brokers: ["broker:9092"]
  
jobs:
  num_pollers: 1
  pipeline_size: 100
  pool:
    num_workers: 1
    max_jobs: 0
    allocate_timeout: 60s
    destroy_timeout: 60s
  pipelines:
    kafka:
      driver: kafka
      config:
        priority: 1
        auto_create_topics_enable: true
        consumer_options:
          topics: [ "feeds", "feeds.feeds.unit", "^[a-zA-Z0-9._-]+$" ]
          consume_regexp: true
        group_options:
          group_id: connect-debezium

consume:
  [
    "kafka"
  ]
    
logs:
  level: debug
  output: stdout

I have a sink connector which sinks the messages with elasticsearch so I know that kafka is working. I can see the messages through the kafka-ui

Nothing is reaching the kafka driver in roadrunner.

I literally have no more clue what else to do to get this to work? I have followed all your instructions I think very well.

I really want to use this product in my project but if I cannot get it to work or get help to get it to work then I don't think it's ready for general use.

3 replies

rustatian Dec 21, 2023
Maintainer

consume section should be at the same level as the pipeline section.

version: '3'

rpc:
  listen: tcp://127.0.0.1:6001

server:
  command: "php consumer.php"

#http:
#  address: "0.0.0.0:8080"
kafka:
  brokers: ["broker:9092"]
  
jobs:
  num_pollers: 1
  pipeline_size: 100
  pool:
    num_workers: 1
    max_jobs: 0
    allocate_timeout: 60s
    destroy_timeout: 60s

  consume: ["kafka"] # <--
  pipelines:
    kafka:
      driver: kafka
      config:
        priority: 1
        auto_create_topics_enable: true
        consumer_options:
          topics: [ "feeds", "feeds.feeds.unit", "^[a-zA-Z0-9._-]+$" ]
          consume_regexp: true
        group_options:
          group_id: connect-debezium

    
logs:
  level: debug
  output: stdout

suityou01 Dec 21, 2023
Author

This is difficult to tell from the example as there is no yaml node folding.

rustatian Dec 21, 2023
Maintainer

I understand you, YAML config is not a very good to read, but in the documentation, you may find a smaller example: https://roadrunner.dev/docs/queues-overview/current/en#common-configuration

suityou01 · 2023-12-21T15:41:29Z

suityou01
Dec 21, 2023
Author

An error message!

2023-12-21T15:38:32+0000	DEBUG	rpc         	plugin was started	{"address": "tcp://127.0.0.1:6001", "list of the plugins with RPC methods:": ["lock", "app", "informer", "jobs", "resetter"]}
2023-12-21T15:38:32+0000	DEBUG	jobs        	initializing driver	{"pipeline": "kafka", "driver": "kafka"}
2023-12-21T15:38:32+0000	DEBUG	kafka       	ping kafka: ok	{"driver": "kafka", "pipeline": "kafka"}
2023-12-21T15:38:32+0000	DEBUG	jobs        	driver ready	{"pipeline": "kafka", "driver": "kafka", "start": "2023-12-21T15:38:32+0000", "elapsed": "5.560543ms"}
2023-12-21T15:38:32+0000	DEBUG	kafka       	pipeline was started	{"driver": "kafka", "pipeline": "kafka", "start": "2023-12-21T15:38:32+0000", "elapsed": "3.234µs"}
2023-12-21T15:38:32+0000	ERROR	kafka       	non-retriable consumer error	{"topic": "", "partition": 0, "code": 23, "description": "The group member's supported protocols are incompatible with those of existing members or first group member tried to join with empty protocol type or empty protocol list.", "message": "INCONSISTENT_GROUP_PROTOCOL"}
2023-12-21T15:38:32+0000	DEBUG	kafka       	kafka listener stopped
2023-12-21T15:38:32+0000	ERROR	kafka       	listener error	{"error": "unable to join group session: INCONSISTENT_GROUP_PROTOCOL: The group member's supported protocols are incompatible with those of existing members or first group member tried to join with empty protocol type or empty protocol list."}
2023-12-21T15:38:32+0000	DEBUG	kafka       	pipeline was stopped	{"driver": "kafka", "pipeline": "kafka", "start": "2023-12-21T15:38:32+0000", "elapsed": "3.450471ms"}
2023-12-21T15:38:32+0000	DEBUG	server      	worker is allocated	{"pid": 16, "internal_event_name": "EventWorkerConstruct"}
2023-12-21T15:38:32+0000	DEBUG	jobs        	exited from jobs pipeline processor
2023-12-21T15:38:32+0000	DEBUG	jobs        	exited from jobs pipeline processor
2023-12-21T15:38:32+0000	DEBUG	jobs        	exited from jobs pipeline processor
2023-12-21T15:38:32+0000	DEBUG	jobs        	exited from jobs pipeline processor
2023-12-21T15:38:32+0000	DEBUG	jobs        	exited from jobs pipeline processor
2023-12-21T15:38:32+0000	DEBUG	jobs        	exited from jobs pipeline processor
2023-12-21T15:38:32+0000	DEBUG	jobs        	exited from jobs pipeline processor
2023-12-21T15:38:32+0000	DEBUG	jobs        	exited from jobs pipeline processor
2023-12-21T15:38:32+0000	DEBUG	jobs        	exited from jobs pipeline processor
2023-12-21T15:38:32+0000	DEBUG	jobs        	exited from jobs pipeline processor
[INFO] RoadRunner server started; version: 2023.3.8, buildtime: 2023-12-14T16:05:26+0000
[INFO] sdnotify: not notified

Process docker container logs smart_planning_engine finished

3 replies

suityou01 Dec 21, 2023
Author

https://forum.confluent.io/t/inconsistent-group-protocol-exception-why/8387

rustatian Dec 21, 2023
Maintainer

@suityou01 After updating group_id, do you have everything running ok?

rustatian Dec 26, 2023
Maintainer

@suityou01 Friendly ping 😃

RoadRunner

How to configure kafka job queue #1820

Uh oh!

Uh oh!

suityou01 Dec 21, 2023

Replies: 9 comments · 14 replies

Uh oh!

Uh oh!

rustatian Dec 21, 2023 Maintainer

Uh oh!

Uh oh!

suityou01 Dec 21, 2023 Author

Uh oh!

rustatian Dec 21, 2023 Maintainer

Uh oh!

msmakouz Dec 21, 2023

Uh oh!

suityou01 Dec 21, 2023 Author

Uh oh!

suityou01 Dec 21, 2023 Author

Uh oh!

rustatian Dec 21, 2023 Maintainer

Uh oh!

suityou01 Dec 21, 2023 Author

Uh oh!

suityou01 Dec 21, 2023 Author

Uh oh!

suityou01 Dec 21, 2023 Author

Uh oh!

rustatian Dec 21, 2023 Maintainer

Uh oh!

rustatian Dec 21, 2023 Maintainer

Uh oh!

Uh oh!

rustatian Dec 21, 2023 Maintainer

Uh oh!

suityou01 Dec 21, 2023 Author

Uh oh!

rustatian Dec 21, 2023 Maintainer

Uh oh!

suityou01 Dec 21, 2023 Author

Uh oh!

rustatian Dec 21, 2023 Maintainer

Uh oh!

suityou01 Dec 21, 2023 Author

Uh oh!

rustatian Dec 21, 2023 Maintainer

Uh oh!

suityou01 Dec 21, 2023 Author

Uh oh!

suityou01 Dec 21, 2023 Author

Uh oh!

rustatian Dec 21, 2023 Maintainer

Uh oh!

rustatian Dec 26, 2023 Maintainer

suityou01
Dec 21, 2023

Replies: 9 comments 14 replies

rustatian
Dec 21, 2023
Maintainer

suityou01
Dec 21, 2023
Author

rustatian Dec 21, 2023
Maintainer

suityou01
Dec 21, 2023
Author

suityou01
Dec 21, 2023
Author

rustatian Dec 21, 2023
Maintainer

suityou01
Dec 21, 2023
Author

suityou01
Dec 21, 2023
Author

suityou01
Dec 21, 2023
Author

rustatian Dec 21, 2023
Maintainer

rustatian Dec 21, 2023
Maintainer

rustatian Dec 21, 2023
Maintainer

suityou01 Dec 21, 2023
Author

rustatian Dec 21, 2023
Maintainer

suityou01
Dec 21, 2023
Author

rustatian Dec 21, 2023
Maintainer

suityou01 Dec 21, 2023
Author

rustatian Dec 21, 2023
Maintainer

suityou01
Dec 21, 2023
Author

suityou01 Dec 21, 2023
Author

rustatian Dec 21, 2023
Maintainer

rustatian Dec 26, 2023
Maintainer