External Kafka queue, Roadrunner, Temporal workflows

[LeonidDereviankoCasafari]

Hi, i'm new to temporal and roadrunner and having some issues with it.
I have local environment started with docker-compose based on example from https://github.com/temporalio/samples-php/blob/master/docker-compose.yml
i take elasticsearch, 'postgresql', 'temporal', 'temporal-admin-tools', 'temporal-ui' from the example and also there is my own app container which is running consumer.
worker.php and saga:consumer cli command based on BookingSaga sample.
I'm trying to run SagaWorkflow for each message consumed from Kafka queue but it won't take any message from Kafka with config i wrote.
If i'm trying to consume messages using Roadrunner When i start and open temporal-ui page i can't see any messages consumed and no workflows started.
But if i use in saga:consumer command consuming messages with RdKafka\KafkaConsumer i can see that there is Workflows in temporal but they are in "Running" state but no workers started, and to complete workers and activities i have to additionally run saga:consumer command in container and only than workflows passing to completed.

So there is two questions:

• How to manage correct consuming from Kafka (the messages should be published by thirdparty service i can't change it) by Roadrunner and start temporal's SagaWorkflow
• How to start workers running automatically with temporal started

It will be excellent it there is example for such combo but i didn't found it

here is rr.yaml and installed requirements with composer

"require": {
        "symfony/dotenv": "^6.2",
        "symfony/console": "^6.3",
        "php": ">=8.2.0",
        "ext-curl": "*",
        "spiral/roadrunner": "^2023.3",
        "spiral/roadrunner-cli": "^2.5",
        "spiral/roadrunner-worker": "^3.2",
        "spiral/roadrunner-jobs": "^4.3",
        "spiral/tokenizer": "^3.9",
        "grpc/grpc": "^1.38",
        "google/protobuf": "^3.25",
        "jobcloud/php-kafka-lib": "^2.0",
        "temporal/sdk": "2.7.x-dev",
        "temporal/open-telemetry-interceptors": "dev-master",
        "guzzlehttp/promises": "^2.0",
        "php-http/httplug": "^2.4",
        "nyholm/psr7": "^1.8",
        "open-telemetry/transport-grpc": "^0.0.17",
        "open-telemetry/exporter-otlp": "^0.0.17"
    },

version: "3"

server:
  relay: pipes
  command: "php /app/app.php saga:consumer queue-name"
  env:
    - TEMPORAL_CLI_ADDRESS: temporal:7233
    - RR_MODE: jobs

temporal:
  address: temporal:7233
  activities:
    num_workers: 1
    allocate_timeout: 3600s
    command: "php /app/worker.php"
    env:
      RR_MODE: temporal

kafka:
  brokers: ["kafka"]

jobs:
  consume: ["kafka-messages"]
  num_pollers: 1
  pipeline_size: 1000
  pool:
    num_workers: 4
    allocate_timeout: 1h
  pipelines:
    kafka-messages:
      driver: kafka
      config:
        brokers: [ kafka:9092 ]
        consumer_options:
          topics: ["queue-name"]
        group_options:
          group_id: "test-consumer"

rpc:
  listen: tcp://127.0.0.1:6001

logs:
  mode: raw
  output: stdout
  err_output: stderr
  encoding: json
  server:
    mode: raw
    level: info
    encoding: json
    output: stdout
    err_output: stderr
  temporal:
    mode: raw
    level: info
    encoding: json
    output: stdout
    err_output: stderr
  rpc:
    mode: raw
    level: info
    encoding: json
    output: stdout
    err_output: stderr

[rustatian]

Hey @LeonidDereviankoCasafari 👋
1.

kafka:
  brokers: ["kafka:9092"] # Port should be used

2. What is your RR version?
3. Have you read this thread: https://github.com/orgs/roadrunner-server/discussions/1730 ?

[LeonidDereviankoCasafari]

Thanks for the answer @rustatian !

i'll check now fix with adding port, and there is a second part of question.
Yes i've read that thread, correct, my queue-name it's a 'topic-name' from Kafka, i missed naming in config in question
but looks like it is a bit different i'm not pushing anything to Kafka but only Read it, and also question related to running workers with temporal in different containers but they are not start automatically.
I'm using RR roadrunner:2023.3.5 version in app container

[rustatian]

Got u 👍
You may read docs here.

Issues in your config:

1. Use number of pollers at least +1 to number of workers.
2. You don't need to set the RR_MODE env variable. RR will set it automatically per pool of workers. By setting it to the RR_MODE: temporal you overwriting mode set by RR for all pools.
3. Simplify logs in the development mode:

logs:
  mode: development
  level: debug

4. Check the docs about kafka. First of all, there is no brokers key. Check that the topic exists on the kafka broker. Or use auto_create_topics_enable pipeline option.

workers with temporal in different containers but they are not start automatically.

Could you please rephrase, I'm not sure that I understand correctly. What is 'different containers' and why they should start automatically?

[LeonidDereviankoCasafari]

About containers i meant that temporal and consumer(with RR on it) are running in different containers, an if i understand correctly it should start something like this: The "App" container is starting and trying to run RR which is connecting to Kafka and looking for temporal-server on "Temporal" container
Next step is "App" is starting saga:consumer (command) script which Register all Workflows and Activities it can find, and with RR trying to get Messages from Kafka and call Workflow using my SagaWFInterface.
But i don't clearly understand which moment Workers are starting: Are they should run on "App" ? if Yes who will start this worker script, "Temporal" through rpc call to "App", or they should ran on "Temporal" starting by RR from App?
this means if i have TWO scripts saga:consumer(app.php) and worker.php where they. should be assign as command in rr.yaml (are the commands assigned in config i show you correctly)
I was looking into that doc's also, The Kafka section. But not sure which config blocks i need in my case at all, i'm not using http, only cli workers
so is it enough to use blocks i show of maybe to add some service or interceptor? or maybe some of them are redundant?

[rustatian]

Workers are PHP-CLI processes started by RR. There are no containers. So, when worker is started, your code inside your file gets executed based on the command you provided.
Here you may find a sample of the RR configuration: link. You may check the repo by the link for the samples of the different PHP workflows.

[LeonidDereviankoCasafari]

Yeah!
looks like i managed it to run properly!

now i have to filter empty messages because of infinite Workflows starting

        $serviceClient = ServiceClient::create('temporal:7233');
        $workflowClient = WorkflowClient::create(
            serviceClient: $serviceClient,
            interceptorProvider: new SimplePipelineProvider([
                new OpenTelemetryWorkflowClientCallsInterceptor(TracerFactory::create('interceptors-sample-client')),
            ])
        );
        $rrConsumer = new Consumer();
            /** @var ReceivedTaskInterface $task */
        while ($task = $rrConsumer->waitTask()) {
            try {
                $name = $task->getName();
                $queue = $task->getQueue();
                $payload = $task->getPayload();

                $workflow = $workflowClient->newWorkflowStub(
                    ConsumerSagaWorkflowInterface::class,
                    WorkflowOptions::new()->withWorkflowExecutionTimeout(CarbonInterval::seconds(180))
                        ->withRetryOptions(
                            RetryOptions::new()
                                ->withInitialInterval(30)
                                ->withMaximumAttempts(2)
                                ->withNonRetryableExceptions([InvalidArgumentException::class]
                                )
                        )
                );
                $run = $workflowClient->start($workflow, $payload);
                $task->complete();
            } catch (Throwable $e) {
                $task->fail($e, requeue: true);
            }
        }

[LeonidDereviankoCasafari]

@rustatian I have containers because i'm running it with Docker, and now i figured out that Workflows keep creating from queue and all workers keep running even when i'm stopped "consumer" container, all is running now in "temporal" container, so RR is sending some command to cache and run workers in temporal.... BUT not for a long time, after some time new Workflows has been created but failed with timeout, so were finished only workers who were already running, but how new Workflows were created? looks like there is some cached queue also.

Thanks for your Help!