Merge pull request #12 from rapideinternet/upgrade/5.7

Update to 5.7 and more

Author: petermein

.styleci.yml

@@ -62,7 +62,6 @@ enabled:
   - print_to_echo
   - self_accessor
   - short_array_syntax
-  - short_scalar_cast
   - single_blank_line_before_namespace
   - single_quote
   - space_after_semicolon
@@ -73,4 +72,4 @@ enabled:
   - unalign_double_arrow
   - unalign_equals
   - unary_operator_spaces
-  - whitespace_after_comma_in_array
+  - whitespace_after_comma_in_array

.travis.yml

@@ -6,18 +6,18 @@ php:
   - 5.6
   - 7.0
   - 7.1
-  - hhvm
+  - 7.2
+  - 7.3
   - nightly
 
 install:
   - ./travis-install.sh
   - travis_retry composer install --no-interaction
 
 matrix:
-    fast_finish: true
-    allow_failures:
-        - php: hhvm
-        - php: nightly
+  fast_finish: true
+  allow_failures:
+    - php: nightly
 
 script:
-  - phpunit --no-coverage
+  - phpunit --no-coverage

README.md

@@ -25,25 +25,45 @@ Kafka Queue driver for Laravel
     $ pecl install rdkafka
     ```
 
-3. Add the following to your php.ini file to enable the php-rdkafka extension
+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`
+	    composer require rapide/laravel-queue-kafka
 
 5. Add LaravelQueueKafkaServiceProvider to `providers` array in `config/app.php`:
 
-	`Rapide\LaravelQueueKafka\LaravelQueueKafkaServiceProvider::class,`
+	    Rapide\LaravelQueueKafka\LaravelQueueKafkaServiceProvider::class,
 
    If you are using Lumen, put this in `bootstrap/app.php`
 
-    `$app->register(Rapide\LaravelQueueKafka\LumenQueueKafkaServiceProvider::class);`
+        $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
@@ -67,4 +87,4 @@ You can contribute to this package by discovering bugs and opening issues. Pleas
 
 #### Supported versions of Laravel 
 
-Tested on: [5.4]
+Tested on: [5.4, 5.5, 5.6, 5.7]

composer.json

@@ -2,6 +2,8 @@
   "name": "rapide/laravel-queue-kafka",
   "description": "Kafka driver for Laravel Queue",
   "license": "MIT",
+  "type": "library",
+  "version": "1.0",
   "authors": [
     {
       "name": "Peter Mein",
@@ -10,9 +12,10 @@
   ],
   "require": {
     "php": ">=5.6.4",
-    "illuminate/database": "5.4.*|5.5.*||5.6.*",
-    "illuminate/support": "5.4.*|5.5.*|5.6.*",
-    "illuminate/queue": "5.4.*|5.5.*|5.6.*"
+    "illuminate/database": "5.4.*|5.5.*||5.6.*|5.7.*",
+    "illuminate/support": "5.4.*|5.5.*|5.6.*|5.7.*",
+    "illuminate/queue": "5.4.*|5.5.*|5.6.*|5.7.*",
+    "ext-rdkafka": "*"
   },
   "require-dev": {
     "phpunit/phpunit": "^5.5",
@@ -23,6 +26,13 @@
       "Rapide\\LaravelQueueKafka\\": "src/"
     }
   },
+  "extra": {
+    "laravel": {
+      "providers": [
+        "Rapide\\LaravelQueueKafka\\LaravelQueueKafkaServiceProvider"
+      ]
+    }
+  },
   "minimum-stability": "dev",
   "prefer-stable": true
 }

config/kafka.php

@@ -32,4 +32,8 @@
      */
     'sleep_on_error' => env('KAFKA_ERROR_SLEEP', 5),
 
+    /*
+     * Sleep when a deadlock is detected
+     */
+    'sleep_on_deadlock' => env('KAFKA_DEADLOCK_SLEEP', 2),
 ];

src/Exceptions/QueueKafkaException.php

@@ -0,0 +1,8 @@
+<?php
+
+namespace Rapide\LaravelQueueKafka\Exceptions;
+
+class QueueKafkaException extends \RuntimeException
+{
+    //
+}

src/Queue/Jobs/KafkaJob.php

@@ -9,15 +9,25 @@
 use Illuminate\Queue\Jobs\Job;
 use Illuminate\Queue\Jobs\JobName;
 use Illuminate\Support\Str;
+use Rapide\LaravelQueueKafka\Exceptions\QueueKafkaException;
 use Rapide\LaravelQueueKafka\Queue\KafkaQueue;
 use RdKafka\Message;
 
 class KafkaJob extends Job implements JobContract
 {
     use DetectsDeadlocks;
 
+    /**
+     * @var KafkaQueue
+     */
     protected $connection;
+    /**
+     * @var KafkaQueue
+     */
     protected $queue;
+    /**
+     * @var Message
+     */
     protected $message;
 
     /**
@@ -55,7 +65,7 @@ public function fire()
                 $this->causedByDeadlock($exception) ||
                 Str::contains($exception->getMessage(), ['detected deadlock'])
             ) {
-                sleep(2);
+                sleep($this->connection->getConfig()['sleep_on_deadlock']);
                 $this->fire();
 
                 return;
@@ -90,8 +100,12 @@ public function getRawBody()
      */
     public function delete()
     {
-        parent::delete();
-        $this->connection->getConsumer()->commitAsync($this->message);
+        try {
+            parent::delete();
+            $this->connection->getConsumer()->commitAsync($this->message);
+        } catch (\RdKafka\Exception $exception) {
+            throw new QueueKafkaException('Could not delete job from the queue', 0, $exception);
+        }
     }
 
     /**
@@ -134,7 +148,7 @@ public function release($delay = 0)
      */
     public function getJobId()
     {
-        return $this->message->get('correlation_id');
+        return $this->message->key;
     }
 
     /**
@@ -162,10 +176,10 @@ private function unserialize(array $body)
             return unserialize($body['data']['command']);
         } catch (Exception $exception) {
             if (
-                $this->causedByDeadlock($exception) ||
-                Str::contains($exception->getMessage(), ['detected deadlock'])
+                $this->causedByDeadlock($exception)
+                || Str::contains($exception->getMessage(), ['detected deadlock'])
             ) {
-                sleep(2);
+                sleep($this->connection->getConfig()['sleep_on_deadlock']);
 
                 return $this->unserialize($body);
             }

src/Queue/KafkaQueue.php

@@ -7,6 +7,7 @@
 use Illuminate\Contracts\Queue\Queue as QueueContract;
 use Illuminate\Queue\Queue;
 use Log;
+use Rapide\LaravelQueueKafka\Exceptions\QueueKafkaException;
 use Rapide\LaravelQueueKafka\Queue\Jobs\KafkaJob;
 
 class KafkaQueue extends Queue implements QueueContract
@@ -19,6 +20,10 @@ class KafkaQueue extends Queue implements QueueContract
      * @var int
      */
     protected $sleepOnError;
+    /**
+     * @var array
+     */
+    protected $config;
     /**
      * @var string
      */
@@ -48,6 +53,7 @@ public function __construct(\RdKafka\Producer $producer, \RdKafka\KafkaConsumer
 
         $this->producer = $producer;
         $this->consumer = $consumer;
+        $this->config = $config;
     }
 
     /**
@@ -84,18 +90,20 @@ public function push($job, $data = '', $queue = null)
      * @param string $queue
      * @param array $options
      *
+     * @throws QueueKafkaException
+     *
      * @return mixed
      */
     public function pushRaw($payload, $queue = null, array $options = [])
     {
         try {
             $topic = $this->getTopic($queue);
 
-            $correlationId = $this->getCorrelationId();
+            $pushRawCorrelationId = $this->getCorrelationId();
 
-            $topic->produce(RD_KAFKA_PARTITION_UA, 0, $payload, $correlationId);
+            $topic->produce(RD_KAFKA_PARTITION_UA, 0, $payload, $pushRawCorrelationId);
 
-            return $correlationId;
+            return $pushRawCorrelationId;
         } catch (ErrorException $exception) {
             $this->reportConnectionError('pushRaw', $exception);
         }
@@ -109,47 +117,54 @@ public function pushRaw($payload, $queue = null, array $options = [])
      * @param mixed $data
      * @param string $queue
      *
+     * @throws QueueKafkaException
+     *
      * @return mixed
      */
     public function later($delay, $job, $data = '', $queue = null)
     {
         //Later is not sup
-        throw new Exception('Later not yet implemented');
+        throw new QueueKafkaException('Later not yet implemented');
     }
 
     /**
      * Pop the next job off of the queue.
      *
      * @param string|null $queue
      *
+     * @throws QueueKafkaException
+     *
      * @return \Illuminate\Queue\Jobs\Job|null
      */
     public function pop($queue = null)
     {
-        $queue = $this->getQueueName($queue);
-        if (!in_array($queue, $this->subscribedQueueNames)) {
-            $this->subscribedQueueNames[] = $queue;
-            $this->consumer->subscribe($this->subscribedQueueNames);
-        }
+        try {
+            $queue = $this->getQueueName($queue);
+            if (!in_array($queue, $this->subscribedQueueNames)) {
+                $this->subscribedQueueNames[] = $queue;
+                $this->consumer->subscribe($this->subscribedQueueNames);
+            }
 
-        $message = $this->consumer->consume(1000);
+            $message = $this->consumer->consume(1000);
 
-        if ($message === null) {
-            return;
-        }
+            if ($message === null) {
+                return null;
+            }
 
-        switch ($message->err) {
-            case RD_KAFKA_RESP_ERR_NO_ERROR:
-                return new KafkaJob(
-                    $this->container, $this, $message,
-                    $this->connectionName, $queue ?: $this->defaultQueue
-                );
-                break;
-            case RD_KAFKA_RESP_ERR__PARTITION_EOF:
-            case RD_KAFKA_RESP_ERR__TIMED_OUT:
-                break;
-            default:
-                throw new \Exception($message->errstr(), $message->err);
+            switch ($message->err) {
+                case RD_KAFKA_RESP_ERR_NO_ERROR:
+                    return new KafkaJob(
+                        $this->container, $this, $message,
+                        $this->connectionName, $queue ?: $this->defaultQueue
+                    );
+                case RD_KAFKA_RESP_ERR__PARTITION_EOF:
+                case RD_KAFKA_RESP_ERR__TIMED_OUT:
+                    break;
+                default:
+                    throw new QueueKafkaException($message->errstr(), $message->err);
+            }
+        } catch (\RdKafka\Exception $exception) {
+            throw new QueueKafkaException('Could not pop from the queue', 0, $exception);
         }
     }
 
@@ -195,6 +210,14 @@ public function getCorrelationId()
         return $this->correlationId ?: uniqid('', true);
     }
 
+    /**
+     * @return array
+     */
+    public function getConfig()
+    {
+        return $this->config;
+    }
+
     /**
      * Create a payload array from the given job and data.
      *
@@ -216,15 +239,15 @@ protected function createPayloadArray($job, $data = '', $queue = null)
      * @param string $action
      * @param Exception $e
      *
-     * @throws Exception
+     * @throws QueueKafkaException
      */
     protected function reportConnectionError($action, Exception $e)
     {
         Log::error('Kafka error while attempting ' . $action . ': ' . $e->getMessage());
 
         // If it's set to false, throw an error rather than waiting
         if ($this->sleepOnError === false) {
-            throw new \RuntimeException('Error writing data to the connection with Kafka');
+            throw new QueueKafkaException('Error writing data to the connection with Kafka');
         }
 
         // Sleep so that we don't flood the log file