In PhpDumper.php line 1561:
[Symfony\Component\DependencyInjection\Exception\InvalidArgumentException]
You cannot dump a container with parameters that contain references to other services (reference to service "App\Messenger\StatsCallBack" found in "/0/stats_cb").
Exception trace:
at /srv/app/vendor/symfony/dependency-injection/Dumper/PhpDumper.php:1561
Symfony\Component\DependencyInjection\Dumper\PhpDumper->exportParameters() at /srv/app/vendor/symfony/dependency-injection/Dumper/PhpDumper.php:1553
Symfony\Component\DependencyInjection\Dumper\PhpDumper->exportParameters() at /srv/app/vendor/symfony/dependency-injection/Dumper/PhpDumper.php:1439
Symfony\Component\DependencyInjection\Dumper\PhpDumper->addDefaultParametersMethod() at /srv/app/vendor/symfony/dependency-injection/Dumper/PhpDumper.php:221
Symfony\Component\DependencyInjection\Dumper\PhpDumper->dump() at /srv/app/vendor/symfony/http-kernel/Kernel.php:780
Symfony\Component\HttpKernel\Kernel->dumpContainer() at /srv/app/vendor/symfony/http-kernel/Kernel.php:606
Symfony\Component\HttpKernel\Kernel->initializeContainer() at /srv/app/vendor/symfony/http-kernel/Kernel.php:136
Symfony\Component\HttpKernel\Kernel->boot() at /srv/app/vendor/symfony/framework-bundle/Console/Application.php:169
Symfony\Bundle\FrameworkBundle\Console\Application->registerCommands() at /srv/app/vendor/symfony/framework-bundle/Console/Application.php:75
Symfony\Bundle\FrameworkBundle\Console\Application->doRun() at /srv/app/vendor/symfony/console/Application.php:149
Symfony\Component\Console\Application->run() at /srv/app/bin/console:39
In KafkaTransportFactory.php line 68:
[RdKafka\Exception (-1)]
Property "stats_cb" must be set through dedicated .._set_..() function
Exception trace:
at /srv/app/vendor/koco/messenger-kafka/src/Messenger/KafkaTransportFactory.php:68
RdKafka\Conf->set() at /srv/app/vendor/koco/messenger-kafka/src/Messenger/KafkaTransportFactory.php:68
Koco\Kafka\Messenger\KafkaTransportFactory->createTransport() at /srv/app/vendor/symfony/messenger/Transport/TransportFactory.php:36
Symfony\Component\Messenger\Transport\TransportFactory->createTransport() at /srv/app/var/cache/test/ContainerT9GqRDx/getMessenger_Transport_ProductupdateService.php:13
require() at /srv/app/var/cache/test/ContainerT9GqRDx/srcApp_KernelTestDebugContainer.php:787
ContainerT9GqRDx\srcApp_KernelTestDebugContainer->load() at /srv/app/vendor/symfony/dependency-injection/Container.php:448
Symfony\Component\DependencyInjection\Container->getService() at /srv/app/vendor/symfony/dependency-injection/Argument/ServiceLocator.php:42
Symfony\Component\DependencyInjection\Argument\ServiceLocator->get() at /srv/app/vendor/symfony/messenger/Command/ConsumeMessagesCommand.php:153
Symfony\Component\Messenger\Command\ConsumeMessagesCommand->execute() at /srv/app/vendor/symfony/console/Command/Command.php:255
Symfony\Component\Console\Command\Command->run() at /srv/app/vendor/symfony/console/Application.php:1027
Symfony\Component\Console\Application->doRunCommand() at /srv/app/vendor/symfony/framework-bundle/Console/Application.php:97
Symfony\Bundle\FrameworkBundle\Console\Application->doRunCommand() at /srv/app/vendor/symfony/console/Application.php:273
Symfony\Component\Console\Application->doRun() at /srv/app/vendor/symfony/framework-bundle/Console/Application.php:83
Symfony\Bundle\FrameworkBundle\Console\Application->doRun() at /srv/app/vendor/symfony/console/Application.php:149
Symfony\Component\Console\Application->run() at /srv/app/bin/console:39
Here is my logger:
<?php
namespace App\Messenger;
use Psr\Log\LoggerInterface;
class StatsCallBack
{
private LoggerInterface $logger;
public function __construct(LoggerInterface $logger)
{
$this->logger = $logger;
}
public function __invoke($kafka, $json, $json_len): void
{
/**
* Allow logging statistics
* @see https://github.com/edenhill/librdkafka/blob/master/STATISTICS.md#partitions
* @see https://github.com/edenhill/librdkafka/wiki/Consumer-lag-monitoring
*/
foreach ($kafka->getAssignment() as $partition) {
$this->logger->info(
'Kafka: Stats on {topic}, {partition}',
[
'json_len' => $json_len,
'json' => $json,
'topic' => $partition->getTopic(),
'partition' => $partition->getPartition(),
]
);
}
}
}
I need it to keep an eye on my consumer lag in order to achieve auto-scaling when lag increase.
Versions
rdkafka
rdkafka support => enabled
version => 4.0.3
build date => May 3 2022 14:54:39
librdkafka version (runtime) => 1.5.0
librdkafka version (build) => 1.5.0.255
Just added a PR. Please let me know if I'm heading forward on the right way.
I have no clue how to test it, I'd be happy to add test if you let me know what is the test strategy on this.
Hi, I didn't find anything to log statistics.
I tried to set
statistics.interval.ms
andstats_cb
but got errors.Using
Result on
Using
Result on
Here is my logger:
I need it to keep an eye on my consumer lag in order to achieve auto-scaling when lag increase.
Versions
rdkafka
rdkafka support => enabled version => 4.0.3 build date => May 3 2022 14:54:39 librdkafka version (runtime) => 1.5.0 librdkafka version (build) => 1.5.0.255
koco/messenger-kafka: v0.17.0