rapideinternet / laravel-queue-kafka

Kafka Queue driver for Laravel

Kafka Queue driver for Laravel

Installation

1. Install librdkafka c library

   $ cd /tmp
   $ mkdir librdkafka
   $ cd librdkafka
   $ git clone https://github.com/edenhill/librdkafka.git .
   $ ./configure
   $ make
   $ make install

2. Install the php-rdkafka PECL extension

   $ pecl install rdkafka

3. a. Add the following to your php.ini file to enable the php-rdkafka extension extension=rdkafka.so

   b. Check if rdkafka is installed
   Note: If you want to run this on php-fpm restart your php-fpm first.

   php -i | grep rdkafka
   

   Your output should look something like this

   rdkafka
   rdkafka support => enabled
   librdkafka version (runtime) => 1.0.0-RC2
   librdkafka version (build) => 0.11.4.0
   

4. Install this package via composer using:

    composer require rapide/laravel-queue-kafka
   

5. Add LaravelQueueKafkaServiceProvider to providers array in config/app.php:

    Rapide\LaravelQueueKafka\LaravelQueueKafkaServiceProvider::class,
   

   If you are using Lumen, put this in bootstrap/app.php

    $app->register(Rapide\LaravelQueueKafka\LumenQueueKafkaServiceProvider::class);
   

6. Add these properties to .env with proper values:

    QUEUE_DRIVER=kafka
   

7. If you want to run a worker for a specific consumer group

    export KAFKA_CONSUMER_GROUP_ID="group2" && php artisan queue:work --sleep=3 --tries=3
   

   Explaination of consumergroups can be found in this article http://blog.cloudera.com/blog/2018/05/scalability-of-kafka-messaging-using-consumer-groups/

Usage

Once you completed the configuration you can use Laravel Queue API. If you used other queue drivers you do not need to change anything else. If you do not know how to use Queue API, please refer to the official Laravel documentation: http://laravel.com/docs/queues

Testing

Run the tests with:

vendor/bin/phpunit

Acknowledgement

This library is inspired by laravel-queue-rabbitmq by vyuldashev. And the Kafka implementations by Superbalist be sure to check those out.

Contribution

You can contribute to this package by discovering bugs and opening issues. Please, add to which version of package you create pull request or issue.

Supported versions of Laravel

Tested on: [5.4, 5.5, 5.6, 5.7]

New version is underway

A version for Laravel 6,7,8 is underway. Also supporting the new RdKafka library

laravel-queue-kafka's Issues

When support Laravel 5.*

Hey, Nice project! I want use it in Laravel 5.5, But have some trouble when i ues composer require rapide/laravel-queue-kafka:

Hope you can update this verison.

Are there any working examples?

Can you show us how to consume the jobs from kafka with console command examples? Would like to know how exactly wire this thing to work with kafka job data

How to Write Kafka Classes?

I have a working Kafka server and Laravel client with your package installed and I understand how the Laravel Queue system works but could you provide some working examples for producing Topics, Consumers and Producers as well as any administrative functions, please?

undefined method

I'm getting this error when i try to use your library. I have installed RDKafka C library and the PECL package.
The error:
local.ERROR: Call to undefined method RdKafka\Message::get() {"exception":"[object] (Symfony\Component\Debug\Exception\FatalThrowableError(code: 0): Call to undefined method RdKafka\Message::get() at /home/vagrant/Code/Laravel/vendor/rapide/laravel-queue-kafka/src/Queue/Jobs/KafkaJob.php:137)

Add laravel 5.5 & auto-discovery

To be able to use the package on Laravel 5.5. we should add the new auto discovery functionality to the composer.json.

attempts not working as expected

When using php artisan queue:work --sleep=3 --tries=3 the attempts attribute will not be updated so the job will be retried for an infinite amount of time. If we set --tries to 1, the job will be processed once

How should we fix this?

Is resident production reconnection supported

Is resident production reconnection supported

Error RdKafka\Conf::setDefaultTopicConf() is deprecated

I have followed the installation method as mentioned in the wiki. But when i run the app , it is throwing an error saying "Function RdKafka\Conf::setDefaultTopicConf() is deprecated".

Remote Kafka server

Hi, i've been test your package and it work fine. but i am intended to install kafka on another machine instead of local machine, i dont find any comment in documantion for this.

would you please explain this one ?

thanks

Todo first release

• Implement size to reflect the remaining item on the topic
• Check the release function
• Fix README to reflect actual data
• Implement unit tests
• Setup Travis

Fix Travis CI by branching versions of phpunit

• Fix test implemenation on travisCI
• Create branches for upgrading
• Update the documentation to guide people to the correct version

1.1.0 is missing on packagist

Please, could you upload tag 1.1.0 on packagist?

Thanks

Class 'RdKafka\Producer' not found

I'm getting this error when i try to use your library.
The error is going on registerDependecies() method in the LaravelQueueKafkaServiceProvider.php line 47

php artisan queue:work on multiple queue

I got multiple topics and it only process one queue.

For example:

php artisan queue:work --daemon --tries=1 --delay=5 --queue=default,low kafka
Will NOT process any job

php artisan queue:work --daemon --tries=1 --delay=5 kafka
Will process only the job from the KAFKA_QUEUE env queue

When debugging I saw that the pop function from KafkaQueue.php will be call for every queue but the RD_KAFKA_RESP_ERR__TIMED_OUT error will be returned every time.

Is there something I am missing?

Job is processed many times.

We use this package on server having started multiple threads for 1 queue (numprocs=30 in supervisor). When job is issued all workers start to process it and it causes errors. Is this package prepared to manage multiple workers?

Support for Laravel 6.0

• Add support for 6.0 without breaking versions

Job does not get pushed to queue.

I am trying to push the job to the queue, here is what I get in the output:

%7|1606284122.968|DESTROY|rdkafka#consumer-10| [thrd:app]: Terminating instance (destroy flags none (0x0))
%7|1606284122.968|TERMINATE|rdkafka#consumer-10| [thrd:app]: Terminating consumer group handler
%7|1606284122.968|CLOSE|rdkafka#consumer-10| [thrd:app]: Closing consumer
%7|1606284122.968|CLOSE|rdkafka#consumer-10| [thrd:app]: Waiting for close events
%7|1606284122.968|CGRPOP|rdkafka#consumer-10| [thrd:main]: Group "test-group" received op TERMINATE in state init (join-state init)
%7|1606284122.968|CGRPTERM|rdkafka#consumer-10| [thrd:main]: Terminating group "test-group" in state init with 0 partition(s)
%7|1606284122.968|CLEARASSIGN|rdkafka#consumer-10| [thrd:main]: No current assignment to clear
%7|1606284122.968|DUMP|rdkafka#consumer-10| [thrd:main]: Assignment dump (started_cnt=0, wait_stop_cnt=0)
%7|1606284122.968|DUMP_ALL|rdkafka#consumer-10| [thrd:main]: List with 0 partition(s):
%7|1606284122.968|DUMP_PND|rdkafka#consumer-10| [thrd:main]: List with 0 partition(s):
%7|1606284122.968|DUMP_QRY|rdkafka#consumer-10| [thrd:main]: List with 0 partition(s):
%7|1606284122.968|DUMP_REM|rdkafka#consumer-10| [thrd:main]: List with 0 partition(s):
%7|1606284122.968|ASSIGNDONE|rdkafka#consumer-10| [thrd:main]: Group "test-group": assignment operations done in join-state init (rebalance rejoin=false)
%7|1606284122.968|CGRPSTATE|rdkafka#consumer-10| [thrd:main]: Group "test-group" changed state init -> term (join-state init)
%7|1606284122.968|BROADCAST|rdkafka#consumer-10| [thrd:main]: Broadcasting state change
%7|1606284122.968|ASSIGNMENT|rdkafka#consumer-10| [thrd:main]: Group "test-group": clearing group assignment
%7|1606284122.968|CLOSE|rdkafka#consumer-10| [thrd:app]: Consumer closed
%7|1606284122.968|TERMINATE|rdkafka#consumer-10| [thrd:app]: Interrupting timers
%7|1606284122.968|TERMINATE|rdkafka#consumer-10| [thrd:app]: Sending TERMINATE to internal main thread
%7|1606284122.968|TERMINATE|rdkafka#consumer-10| [thrd:app]: Joining internal main thread
%7|1606284122.968|TERMINATE|rdkafka#consumer-10| [thrd:main]: Internal main thread terminating
%7|1606284122.968|DESTROY|rdkafka#consumer-10| [thrd:main]: Destroy internal
%7|1606284122.968|BROADCAST|rdkafka#consumer-10| [thrd:main]: Broadcasting state change
%7|1606284122.968|DESTROY|rdkafka#consumer-10| [thrd:main]: Removing all topics
%7|1606284122.968|DESTROY|rdkafka#consumer-10| [thrd:main]: Sending TERMINATE to [47.245.57.245:9094/bootstrap
%7|1606284122.968|DESTROY|rdkafka#consumer-10| [thrd:main]: Sending TERMINATE to GroupCoordinator
%7|1606284122.968|TERM|rdkafka#consumer-10| [thrd:[47.245.57.245:9094/bootstrap]: [47.245.57.245:9094/bootstrap: Received TERMINATE op in state INIT: 1 refcnts, 0 toppar(s), 0 active toppar(s), 0 outbufs, 0 waitresps, 0 retrybufs
%7|1606284122.968|TERM|rdkafka#consumer-10| [thrd:GroupCoordinator]: GroupCoordinator: Received TERMINATE op in state INIT: 1 refcnts, 0 toppar(s), 0 active toppar(s), 0 outbufs, 0 waitresps, 0 retrybufs
%7|1606284122.968|FAIL|rdkafka#consumer-10| [thrd:[47.245.57.245:9094/bootstrap]: [47.245.57.245:9094/bootstrap: Client is terminating (after 2ms in state INIT) (_DESTROY)
%7|1606284122.968|STATE|rdkafka#consumer-10| [thrd:[47.245.57.245:9094/bootstrap]: [47.245.57.245:9094/bootstrap: Broker changed state INIT -> DOWN
%7|1606284122.968|BROADCAST|rdkafka#consumer-10| [thrd:[47.245.57.245:9094/bootstrap]: Broadcasting state change
%7|1606284122.968|BUFQ|rdkafka#consumer-10| [thrd:[47.245.57.245:9094/bootstrap]: [47.245.57.245:9094/bootstrap: Purging bufq with 0 buffers
%7|1606284122.968|BUFQ|rdkafka#consumer-10| [thrd:[47.245.57.245:9094/bootstrap]: [47.245.57.245:9094/bootstrap: Purging bufq with 0 buffers
%7|1606284122.968|BUFQ|rdkafka#consumer-10| [thrd:[47.245.57.245:9094/bootstrap]: [47.245.57.245:9094/bootstrap: Updating 0 buffers on connection reset
%7|1606284122.968|FAIL|rdkafka#consumer-10| [thrd:GroupCoordinator]: GroupCoordinator: Client is terminating (after 2ms in state INIT) (_DESTROY)
%7|1606284122.968|STATE|rdkafka#consumer-10| [thrd:GroupCoordinator]: GroupCoordinator: Broker changed state INIT -> DOWN
%7|1606284122.968|BROADCAST|rdkafka#consumer-10| [thrd:GroupCoordinator]: Broadcasting state change
%7|1606284122.968|BUFQ|rdkafka#consumer-10| [thrd:GroupCoordinator]: GroupCoordinator: Purging bufq with 0 buffers
%7|1606284122.968|BUFQ|rdkafka#consumer-10| [thrd:GroupCoordinator]: GroupCoordinator: Purging bufq with 0 buffers
%7|1606284122.968|BUFQ|rdkafka#consumer-10| [thrd:GroupCoordinator]: GroupCoordinator: Updating 0 buffers on connection reset
%7|1606284122.968|TERMINATE|rdkafka#consumer-10| [thrd:[47.245.57.245:9094/bootstrap]: [47.245.57.245:9094/bootstrap: Handle is terminating in state DOWN: 1 refcnts (0x7ffcaf822d98), 0 toppar(s), 0 active toppar(s), 0 outbufs, 0 waitresps, 0 retrybufs: failed 0 request(s) in retry+outbuf
%7|1606284122.968|FAIL|rdkafka#consumer-10| [thrd:[47.245.57.245:9094/bootstrap]: [47.245.57.245:9094/bootstrap: Broker handle is terminating (after 0ms in state DOWN) (_DESTROY)
%7|1606284122.968|BUFQ|rdkafka#consumer-10| [thrd:[47.245.57.245:9094/bootstrap]: [47.245.57.245:9094/bootstrap: Purging bufq with 0 buffers
%7|1606284122.968|BUFQ|rdkafka#consumer-10| [thrd:[47.245.57.245:9094/bootstrap]: [47.245.57.245:9094/bootstrap: Purging bufq with 0 buffers
%7|1606284122.968|TERMINATE|rdkafka#consumer-10| [thrd:main]: Purging reply queue
%7|1606284122.968|TERMINATE|rdkafka#consumer-10| [thrd:GroupCoordinator]: GroupCoordinator: Handle is terminating in state DOWN: 1 refcnts (0x7ffcb0014798), 0 toppar(s), 0 active toppar(s), 0 outbufs, 0 waitresps, 0 retrybufs: failed 0 request(s) in retry+outbuf
%7|1606284122.968|TERMINATE|rdkafka#consumer-10| [thrd:main]: Decommissioning internal broker
%7|1606284122.968|BUFQ|rdkafka#consumer-10| [thrd:[47.245.57.245:9094/bootstrap]: [47.245.57.245:9094/bootstrap: Updating 0 buffers on connection reset
%7|1606284122.968|FAIL|rdkafka#consumer-10| [thrd:GroupCoordinator]: GroupCoordinator: Broker handle is terminating (after 0ms in state DOWN) (_DESTROY)
%7|1606284122.968|BUFQ|rdkafka#consumer-10| [thrd:GroupCoordinator]: GroupCoordinator: Purging bufq with 0 buffers
%7|1606284122.968|BUFQ|rdkafka#consumer-10| [thrd:GroupCoordinator]: GroupCoordinator: Purging bufq with 0 buffers
%7|1606284122.968|BUFQ|rdkafka#consumer-10| [thrd:GroupCoordinator]: GroupCoordinator: Updating 0 buffers on connection reset
%7|1606284122.968|TERM|rdkafka#consumer-10| [thrd::0/internal]: :0/internal: Received TERMINATE op in state INIT: 1 refcnts, 0 toppar(s), 0 active toppar(s), 0 outbufs, 0 waitresps, 0 retrybufs
%7|1606284122.968|FAIL|rdkafka#consumer-10| [thrd::0/internal]: :0/internal: Client is terminating (after 2ms in state INIT) (_DESTROY)
%7|1606284122.968|STATE|rdkafka#consumer-10| [thrd::0/internal]: :0/internal: Broker changed state INIT -> DOWN
%7|1606284122.968|BROADCAST|rdkafka#consumer-10| [thrd::0/internal]: Broadcasting state change
%7|1606284122.968|BUFQ|rdkafka#consumer-10| [thrd::0/internal]: :0/internal: Purging bufq with 0 buffers
%7|1606284122.968|BUFQ|rdkafka#consumer-10| [thrd::0/internal]: :0/internal: Purging bufq with 0 buffers
%7|1606284122.968|BUFQ|rdkafka#consumer-10| [thrd::0/internal]: :0/internal: Updating 0 buffers on connection reset
%7|1606284122.968|TERMINATE|rdkafka#consumer-10| [thrd:main]: Join 3 broker thread(s)
%7|1606284122.968|TERMINATE|rdkafka#consumer-10| [thrd::0/internal]: :0/internal: Handle is terminating in state DOWN: 1 refcnts (0x7ffcaf823f98), 0 toppar(s), 0 active toppar(s), 0 outbufs, 0 waitresps, 0 retrybufs: failed 0 request(s) in retry+outbuf
%7|1606284122.968|FAIL|rdkafka#consumer-10| [thrd::0/internal]: :0/internal: Broker handle is terminating (after 0ms in state DOWN) (_DESTROY)
%7|1606284122.968|BUFQ|rdkafka#consumer-10| [thrd::0/internal]: :0/internal: Purging bufq with 0 buffers
%7|1606284122.968|BUFQ|rdkafka#consumer-10| [thrd::0/internal]: :0/internal: Purging bufq with 0 buffers
%7|1606284122.968|BUFQ|rdkafka#consumer-10| [thrd::0/internal]: :0/internal: Updating 0 buffers on connection reset
%7|1606284122.968|TERMINATE|rdkafka#consumer-10| [thrd:main]: Internal main thread termination done
%7|1606284122.968|TERMINATE|rdkafka#consumer-10| [thrd:app]: Destroying op queues
%7|1606284122.968|TERMINATE|rdkafka#consumer-10| [thrd:app]: Destroying cgrp
%7|1606284122.968|MEMBERID|rdkafka#consumer-10| [thrd:app]: Group "test-group": updating member id "" -> "(not-set)"
%7|1606284122.968|TERMINATE|rdkafka#consumer-10| [thrd:app]: Termination done: freeing resources
%4|1606284122.968|TERMINATE|rdkafka#producer-9| [thrd:app]: Producer terminating with 1 message (541 bytes) still in queue or transit: use flush() to wait for outstanding message delivery

but the queue is empty.

Error when i run "export KAFKA_CONSUMER_GROUP_ID="group2" && php artisan queue:work --sleep=3 --tries=3"

In kafka.php I have specified the broker

'brokers' => env('KAFKA_BROKERS', '172.105.23.57:9092').

This is to connect ::(KafkaConnector.php)

    $producer = $this->container->makeWith('queue.kafka.producer', []);
    $producer->addBrokers('172.105.23.57:9092');

    $topicConf = $this->container->makeWith('queue.kafka.topic_conf', []);
    $topicConf->set('auto.offset.reset', 'largest');

    $conf = $this->container->makeWith('queue.kafka.conf', []);
    $conf->set('group.id', array_get($config, 'consumer_group_id', 'php-pubsub'));
    $conf->set('metadata.broker.list', '172.105.23.57:9092');
  
    $consumer = $this->container->makeWith('queue.kafka.consumer', ['conf' => $conf]);

    return new KafkaQueue(
        $producer,
        $consumer,
        $config
    );

Error::

%3|1574254453.078|FAIL|rdkafka#producer-1| [thrd:172.105.23.57:9092/bootstrap]: 172.105.23.57:9092/bootstrap: Connect to ipv4#172.105.23.57:9092 failed: Connect ion refused (after 1028ms in state CONNECT)
%3|1574254453.078|ERROR|rdkafka#producer-1| [thrd:172.105.23.57:9092/bootstrap]: 172.105.23.57:9092/bootstrap: Connect to ipv4#172.105.23.57:9092 failed: Connec tion refused (after 1028ms in state CONNECT)
%3|1574254453.078|ERROR|rdkafka#producer-1| [thrd:172.105.23.57:9092/bootstrap]: 1/1 brokers are down

And I have already changed advertisers.listeners in server.properties.