[v1.8.x] Add restart command

src/Config/Config.php

@@ -78,6 +78,7 @@ public function __construct(
         private array              $customOptions = [],
         ?HandlesBatchConfiguration $batchConfig = null,
         private bool               $stopAfterLastMessage = false,
+        private int                $restartInterval = 1000,
     ) {
         $this->batchConfig = $batchConfig ?? new NullBatchConfig();
     }
@@ -160,6 +161,11 @@ public function getBatchConfig(): HandlesBatchConfiguration
         return $this->batchConfig;
     }
 
+    public function getRestartInterval(): int
+    {
+        return $this->restartInterval;
+    }
+
     #[Pure]
     private function getSaslOptions(): array
     {

src/Console/Commands/KafkaRestartConsumersCommand.php

@@ -0,0 +1,22 @@
+<?php
+
+namespace Junges\Kafka\Console\Commands;
+
+use Illuminate\Console\Command;
+use Illuminate\Support\Facades\Cache;
+use Illuminate\Support\InteractsWithTime;
+
+class KafkaRestartConsumersCommand extends Command
+{
+    use InteractsWithTime;
+
+    protected $signature = 'kafka:restart-consumers';
+
+    protected $description = 'Restart all Kafka consumers.';
+
+    public function handle()
+    {
+        Cache::forever('laravel-kafka:consumer:restart', $this->currentTime());
+        $this->info('Kafka consumers restart signal sent.');
+    }
+}

src/Consumers/Consumer.php

@@ -3,25 +3,27 @@
 namespace Junges\Kafka\Consumers;
 
 use Closure;
+use Throwable;
+use RdKafka\Conf;
+use RdKafka\Message;
+use Junges\Kafka\Logger;
+use RdKafka\KafkaConsumer;
+use Junges\Kafka\Retryable;
+use Junges\Kafka\Config\Config;
+use Junges\Kafka\MessageCounter;
 use Illuminate\Support\Collection;
-use Junges\Kafka\Commit\Contracts\Committer;
-use Junges\Kafka\Commit\Contracts\CommitterFactory;
-use Junges\Kafka\Commit\DefaultCommitterFactory;
+use Illuminate\Support\Facades\Cache;
 use Junges\Kafka\Commit\NativeSleeper;
-use Junges\Kafka\Config\Config;
+use RdKafka\Producer as KafkaProducer;
+use Junges\Kafka\Message\ConsumedMessage;
+use Junges\Kafka\Commit\Contracts\Committer;
 use Junges\Kafka\Contracts\CanConsumeMessages;
-use Junges\Kafka\Contracts\KafkaConsumerMessage;
 use Junges\Kafka\Contracts\MessageDeserializer;
+use Junges\Kafka\Commit\DefaultCommitterFactory;
+use Junges\Kafka\Contracts\KafkaConsumerMessage;
+use Junges\Kafka\Commit\Contracts\CommitterFactory;
 use Junges\Kafka\Exceptions\KafkaConsumerException;
-use Junges\Kafka\Logger;
-use Junges\Kafka\Message\ConsumedMessage;
-use Junges\Kafka\MessageCounter;
-use Junges\Kafka\Retryable;
-use RdKafka\Conf;
-use RdKafka\KafkaConsumer;
-use RdKafka\Message;
-use RdKafka\Producer as KafkaProducer;
-use Throwable;
+use Junges\Kafka\Support\Timer;
 
 class Consumer implements CanConsumeMessages
 {
@@ -50,6 +52,8 @@ class Consumer implements CanConsumeMessages
     private MessageDeserializer $deserializer;
     private bool $stopRequested = false;
     private ?Closure $onStopConsume = null;
+    protected int $lastRestart = 0;
+    protected Timer $restartTimer;
 
     /**
      * @param \Junges\Kafka\Config\Config $config
@@ -74,6 +78,8 @@ public function __construct(private Config $config, MessageDeserializer $deseria
     public function consume(): void
     {
         $this->cancelStopConsume();
+        $this->configureRestartTimer();
+
         $this->consumer = app(KafkaConsumer::class, [
             'conf' => $this->setConf($this->config->getConsumerOptions()),
         ]);
@@ -92,6 +98,7 @@ public function consume(): void
 
         do {
             $this->retryable->retry(fn () => $this->doConsume());
+            $this->checkForRestart();
         } while (! $this->maxMessagesLimitReached() && ! $this->stopRequested);
 
         if ($this->onStopConsume) {
@@ -369,4 +376,28 @@ private function getConsumerMessage(Message $message): KafkaConsumerMessage
             'timestamp' => $message->timestamp,
         ]);
     }
+
+    protected function configureRestartTimer(): void
+    {
+        $this->lastRestart = $this->getLastRestart();
+        $this->restartTimer = new Timer();
+        $this->restartTimer->start($this->config->getRestartInterval());
+    }
+
+    protected function checkForRestart(): void
+    {
+        if(!$this->restartTimer->isTimedOut()) {
+            return;
+        }
+
+        $this->restartTimer->start($this->config->getRestartInterval());
+        if ($this->lastRestart !== $this->getLastRestart()) {
+            $this->stopRequested = true;
+        }
+    }
+
+    protected function getLastRestart(): int
+    {
+        return Cache::get('laravel-kafka:consumer:restart', 0);
+    }
 }

src/Providers/LaravelKafkaServiceProvider.php

@@ -2,16 +2,17 @@
 
 namespace Junges\Kafka\Providers;
 
+use Junges\Kafka\Message\Message;
 use Illuminate\Support\ServiceProvider;
-use Junges\Kafka\Console\Commands\KafkaConsumerCommand;
+use Junges\Kafka\Message\ConsumedMessage;
+use Junges\Kafka\Contracts\MessageSerializer;
+use Junges\Kafka\Contracts\MessageDeserializer;
 use Junges\Kafka\Contracts\KafkaConsumerMessage;
 use Junges\Kafka\Contracts\KafkaProducerMessage;
-use Junges\Kafka\Contracts\MessageDeserializer;
-use Junges\Kafka\Contracts\MessageSerializer;
-use Junges\Kafka\Message\ConsumedMessage;
-use Junges\Kafka\Message\Deserializers\JsonDeserializer;
-use Junges\Kafka\Message\Message;
 use Junges\Kafka\Message\Serializers\JsonSerializer;
+use Junges\Kafka\Console\Commands\KafkaConsumerCommand;
+use Junges\Kafka\Message\Deserializers\JsonDeserializer;
+use Junges\Kafka\Console\Commands\KafkaRestartConsumersCommand;
 
 class LaravelKafkaServiceProvider extends ServiceProvider
 {
@@ -22,6 +23,7 @@ public function boot(): void
         if ($this->app->runningInConsole()) {
             $this->commands([
                 KafkaConsumerCommand::class,
+                KafkaRestartConsumersCommand::class
             ]);
         }
     }

tests/Consumers/ConsumerTest.php

@@ -2,19 +2,20 @@
 
 namespace Junges\Kafka\Tests\Consumers;
 
-use Junges\Kafka\Commit\Contracts\CommitterFactory;
-use Junges\Kafka\Commit\VoidCommitter;
+use RdKafka\Message;
 use Junges\Kafka\Config\Config;
-use Junges\Kafka\Consumers\Consumer;
-use Junges\Kafka\Contracts\KafkaConsumerMessage;
-use Junges\Kafka\Exceptions\KafkaConsumerException;
 use Junges\Kafka\Facades\Kafka;
+use Junges\Kafka\Consumers\Consumer;
+use Junges\Kafka\Commit\VoidCommitter;
 use Junges\Kafka\Message\ConsumedMessage;
-use Junges\Kafka\Message\Deserializers\JsonDeserializer;
-use Junges\Kafka\Tests\Fakes\FakeConsumer;
 use Junges\Kafka\Tests\Fakes\FakeHandler;
+use Junges\Kafka\Tests\Fakes\FakeConsumer;
+use Junges\Kafka\Consumers\CallableConsumer;
 use Junges\Kafka\Tests\LaravelKafkaTestCase;
-use RdKafka\Message;
+use Junges\Kafka\Contracts\KafkaConsumerMessage;
+use Junges\Kafka\Commit\Contracts\CommitterFactory;
+use Junges\Kafka\Exceptions\KafkaConsumerException;
+use Junges\Kafka\Message\Deserializers\JsonDeserializer;
 
 class ConsumerTest extends LaravelKafkaTestCase
 {
@@ -198,4 +199,58 @@ public function testItAcceptsCustomCommitter(): void
         $committer = $this->getPropertyWithReflection('committer', $consumer);
         $this->assertInstanceOf(VoidCommitter::class, $committer);
     }
+
+    public function testItCanRestartConsumer()
+    {
+        $message = new Message();
+        $message->err = 0;
+        $message->key = 'key';
+        $message->topic_name = 'test';
+        $message->payload = '{"body": "message payload"}';
+        $message->offset = 0;
+        $message->partition = 1;
+        $message->headers = [];
+
+        $message2 = new Message();
+        $message2->err = 0;
+        $message2->key = 'key2';
+        $message2->topic_name = 'test';
+        $message2->payload = '{"body": "message payload2"}';
+        $message2->offset = 0;
+        $message2->partition = 1;
+        $message2->headers = [];
+
+        $this->mockConsumerWithMessage($message, $message2);
+        $this->mockProducer();
+
+
+        $fakeHandler = new CallableConsumer(
+            function (KafkaConsumerMessage $message) {
+                    // sleep 100 miliseconds to simulate restart interval check
+                    usleep(100 * 1000);
+                    $this->artisan('kafka:restart-consumers');
+            }, 
+            []
+        );
+
+        $config = new Config(
+            broker: 'broker',
+            topics: ['test-topic'],
+            securityProtocol: 'security',
+            commit: 1,
+            groupId: 'group',
+            consumer: $fakeHandler,
+            sasl: null,
+            dlq: null,
+            maxMessages: 2,
+            maxCommitRetries: 1,
+            restartInterval : 100
+        );
+
+        $consumer = new Consumer($config, new JsonDeserializer());
+        $consumer->consume();
+
+        //finaly only one message should be consumed
+        $this->assertEquals(1, $consumer->consumedMessagesCount());
+    }
 }