RoadRunner
[JOBS]: How correctly connect RR v2023.3 with Kafka broker #1791
-
|
composer.json {
"name": "complayer",
"description": "Compatibility layer for SE Collector with the Saga pattern on Temporal",
"type": "project",
"minimum-stability": "dev",
"prefer-stable": true,
"require": {
"spiral/roadrunner": "^2023.3",
"spiral/roadrunner-jobs": "^4.2.0",
"spiral/roadrunner-worker": "^3.0",
"temporal/sdk": "2.7.x",
"temporal/open-telemetry-interceptors": "dev-master",
"open-telemetry/exporter-otlp": "^0.0.17",
"open-telemetry/transport-grpc": "^0.0.17",
"symfony/dotenv": "^6.2",
"spiral/tokenizer": "^3.8",
"symfony/console": "^6.3",
"google/protobuf": "^3.21",
"grpc/grpc": "^1.42",
"php": ">=8.1.0",
"ext-grpc": "*",
"ext-curl": "*",
"guzzlehttp/promises": "^2.0",
"php-http/httplug": "^2.4",
"nyholm/psr7": "^1.8"
},
"require-dev": {
"phpunit/phpunit": "^9.6.11"
},
"license": "proprietary",
"autoload": {
"psr-4": {
"Collector\\": "grpc/Collector",
"GPBMetadata\\": [
"grpc/GPBMetadata"
],
"Models\\": [
"grpc/Models"
],
"Complayer\\": "src/"
}
},
"config": {
"allow-plugins": {
"php-http/discovery": true
}
}
}sample of configuration roadrunner 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"
command: "/usr/bin/php /app/consumer.php"
kafka:
brokers: ["kafka"]
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"
temporal:
address: temporal:7233
activities:
num_workers: 1
rpc:
listen: tcp://127.0.0.1:6001consumer.php <?php
use Spiral\RoadRunner\Jobs\Task\ReceivedTaskInterface;
use Spiral\RoadRunner\Jobs\Jobs;
use Spiral\RoadRunner;
use Spiral\RoadRunner\Worker;
use Temporal\Client\WorkflowClient;
require_once __DIR__.'/vendor/autoload.php';
$totalEventConsumed = 0;
$consumer = new \Spiral\RoadRunner\Jobs\Consumer();
$shouldBeRestarted = false;
echo "Consumer was created ...";
/** @var ReceivedTaskInterface $task */
while ($task = $consumer->waitTask()) {
try {
var_dump($task);
/*$payload = $task->getPayload();
if(empty($payload['Data'])){
continue;
}
$eventData = json_decode(base64_decode($payload['Data']), true);
$totalEventConsumed++;*/
$task->complete();
} catch (\Throwable $e) {
$task->fail($e, $shouldBeRestarted);
}
}
echo "Events were consumed ...";but I remember that in the previous version |
Beta Was this translation helpful? Give feedback.
Replies: 3 comments 19 replies
-
|
Hey @denyskanunnikov 👋
So, there are no problems with a connection to the Kafka broker. The issue is that you consumed temporal data as JOB message. |
Beta Was this translation helpful? Give feedback.
-
|
@rustatian thank you very much for the explanation! |
Beta Was this translation helpful? Give feedback.
-
|
Just to double-check. You want to send a message to the temporal worker from the JOBS worker, right? |
Beta Was this translation helpful? Give feedback.
-
|
yes, you are right |
Beta Was this translation helpful? Give feedback.
-
|
I'm sure this is possible. Let me call @wolfy-j here, since I'm not a PHP dev 😢 |
Beta Was this translation helpful? Give feedback.
-
|
Hi, You can freely connect to the temporal server from any of the workers (well, except Workflow workers; these workers are special). Implement a handler that communicates with temporal using WorkflowClient. You can start workflows or send signals to running workflow; there are no limitations. |
Beta Was this translation helpful? Give feedback.
-
|
@wolfy-j, thanks. but I have a slightly different design in communication and I wanted to implement consuming of a message from the kafka topic by the roadrunner worker, carry out a simple validation context of this event and then transfer the data to the workflow... |
Beta Was this translation helpful? Give feedback.
-
|
Sure, you can do that. |
Beta Was this translation helpful? Give feedback.
-
|
my issue was solved |
Beta Was this translation helpful? Give feedback.
-
|
No, this env is available via the regular PHP getenv call. RR sets this env per pool. You may use a regular PHP |
Beta Was this translation helpful? Give feedback.
-
|
sorry, but that's a pretty confusing comment for me..... do you mean that on the realisation side of the roadrunner worker I have to check what mode is set? |
Beta Was this translation helpful? Give feedback.
-
|
Yep, you're right 😃 |
Beta Was this translation helpful? Give feedback.
-
|
@rustatian this simple code working correctly. thank you guys very much ! I now fully understand how to configure roadrunner modes and run different processes based on enviroment variables |
Beta Was this translation helpful? Give feedback.
-
|
Sure thing 👍 |
Beta Was this translation helpful? Give feedback.
Hi,
You can freely connect to the temporal server from any of the workers (well, except Workflow workers; these workers are special).
Implement a handler that communicates with temporal using WorkflowClient. You can start workflows or send signals to running workflow; there are no limitations.