🌈 Added better exception handling

🎨 StyleCI

Author: petermein

config/kafka.php

@@ -32,7 +32,7 @@
      */
     '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,10 @@
+<?php
+
+
+namespace Rapide\LaravelQueueKafka\Exceptions;
+
+
+class QueueKafkaException extends \RuntimeException
+{
+
+}

src/Queue/Jobs/KafkaJob.php

@@ -9,6 +9,7 @@
 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;
 
@@ -81,7 +82,7 @@ public function fire()
      */
     public function attempts()
     {
-        return (int)($this->payload()['attempts']) + 1;
+        return (int) ($this->payload()['attempts']) + 1;
     }
 
     /**
@@ -99,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);
+        }
     }
 
     /**
@@ -171,8 +176,8 @@ 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($this->connection->getConfig()['sleep_on_deadlock']);
 

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
@@ -90,17 +91,18 @@ public function push($job, $data = '', $queue = null)
      * @param array $options
      *
      * @return mixed
+     * @throws QueueKafkaException
      */
     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);
         }
@@ -119,7 +121,7 @@ public function pushRaw($payload, $queue = null, array $options = [])
     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');
     }
 
     /**
@@ -128,33 +130,37 @@ public function later($delay, $job, $data = '', $queue = null)
      * @param string|null $queue
      *
      * @return \Illuminate\Queue\Jobs\Job|null
+     * @throws QueueKafkaException
      */
     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);
         }
     }
 
@@ -229,15 +235,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