#93909 add check-offsets and set-offset artisan commands

Author: Ivan Koryukov

composer.json

@@ -26,12 +26,13 @@
     "require-dev": {
         "brianium/paratest": "^6.2",
         "friendsofphp/php-cs-fixer": "^3.2",
+        "kwn/php-rdkafka-stubs": "^2.2",
         "nunomaduro/collision": "^5.3",
         "orchestra/testbench": "^6.15",
         "pestphp/pest": "^1.18",
         "pestphp/pest-plugin-laravel": "^1.1",
-        "phpunit/phpunit": "^9.3",
         "php-parallel-lint/php-var-dump-check": "^0.5.0",
+        "phpunit/phpunit": "^9.3",
         "spatie/laravel-ray": "^1.9"
     },
     "autoload": {

src/Commands/KafkaCheckOffsetsCommand.php

@@ -0,0 +1,51 @@
+<?php
+
+namespace Ensi\LaravelPhpRdKafkaConsumer\Commands;
+
+use Ensi\LaravelPhpRdKafkaConsumer\HighLevelConsumer;
+use Illuminate\Console\Command;
+
+class KafkaCheckOffsetsCommand extends Command
+{
+    protected $signature = 'kafka:check-offsets';
+    protected $description = 'Проверить что для всех используемых топиков задан offset';
+
+    public function handle(): int
+    {
+        $configuredTopics = config('kafka-consumer.processors', []);
+        $success = true;
+        foreach ($configuredTopics as $configuredTopic) {
+            $success &= $this->checkTopic($configuredTopic['consumer'], $configuredTopic['topic']);
+            sleep(1);
+        }
+
+
+        return $success ? self::SUCCESS : self::FAILURE;
+    }
+
+    protected function checkTopic(string $consumerName, string $topicName): bool
+    {
+        $this->getOutput()->writeln("<bg=cyan>Check topic:</> {$topicName}");
+        /** @var HighLevelConsumer $consumer */
+        $consumer = resolve(HighLevelConsumer::class);
+        $consumer->for($consumerName);
+        $partitions = $consumer->getPartitions($topicName);
+        if (!$partitions) {
+            $this->getOutput()->writeln("    <fg=red>Topic {$topicName} doesn't exists!</>");
+            return false;
+        }
+        $success = true;
+        foreach ($partitions as $partition) {
+            if ($partition->getOffset() < 0) {
+                $success = false;
+                $this->getOutput()->writeln("    <fg=red>No offset for partition: {$partition->getPartition()}</>");
+            }
+        }
+
+        if ($success) {
+            $this->getOutput()->writeln("    <fg=green>No errors</>");
+        }
+
+        return $success;
+    }
+}

src/Commands/KafkaSetOffset.php

@@ -0,0 +1,59 @@
+<?php
+
+namespace Ensi\LaravelPhpRdKafkaConsumer\Commands;
+
+use Ensi\LaravelPhpRdKafkaConsumer\HighLevelConsumer;
+use Illuminate\Console\Command;
+use RdKafka\Exception as RdKafkaException;
+
+class KafkaSetOffset extends Command
+{
+    public const EARLIEST = 'earliest';
+    public const LATEST = 'latest';
+
+    protected $signature = 'kafka:set-offset
+                            {--consumer= : name of consumer}
+                            {--topic= : name of topic}
+                            {--partition= : id of partition}
+                            {--offset= : desired offset, "earliest", "latest" or positive integer}';
+    protected $description = "Задать смещение для консюмера в топике";
+
+    public function handle()
+    {
+        $consumerName = $this->option('consumer');
+        $topicName = $this->option('topic');
+        $partitionId = $this->option('partition');
+        $offset = $this->option('offset');
+
+        if (!in_array($offset, [self::EARLIEST, self::LATEST]) && !is_numeric($offset)) {
+            $this->getOutput()->writeln("<fg=red>Error: Invalid offset</>");
+            return self::INVALID;
+        }
+
+        /** @var HighLevelConsumer $consumer */
+        $consumer = resolve(HighLevelConsumer::class);
+        $consumer->for($consumerName);
+
+        try {
+            $bounds = $consumer->getPartitionBounds($topicName, $partitionId);
+            $realOffset = match($offset) {
+                'earliest' => $bounds[0] ?? 0,
+                'latest' => $bounds[1] ?? 0,
+                default => $offset,
+            };
+
+            if ($offset != $realOffset) {
+                $this->getOutput()->writeln("<fg=yellow>Use {$offset} offset:</> {$realOffset}");
+            }
+
+            $consumer->setOffset($topicName, $partitionId, $realOffset);
+        } catch (RdKafkaException $e) {
+            $this->getOutput()->writeln("<fg=red>Error: {$e->getMessage()}</>");
+
+            return self::INVALID;
+        }
+        $this->getOutput()->writeln("<fg=green>Success</>");
+
+        return self::SUCCESS;
+    }
+}

src/HighLevelConsumer.php

@@ -8,6 +8,7 @@
 use RdKafka\Exception as RdKafkaException;
 use RdKafka\KafkaConsumer;
 use RdKafka\Message;
+use RdKafka\TopicPartition;
 use Throwable;
 
 class HighLevelConsumer
@@ -126,4 +127,49 @@ protected function shouldBeStopped(int|float $startTime, int $eventsProcessed, C
 
         return $this->forceStop;
     }
+
+    /**
+     * @param string $topicName
+     * @return TopicPartition[]
+     * @throws RdKafkaException
+     */
+    public function getPartitions(string $topicName): array
+    {
+        $metadata = $this->consumer->getMetadata(true, null, 1000);
+        foreach ($metadata->getTopics() as $topicMeta) {
+            if ($topicMeta->getTopic() != $topicName) {
+                continue;
+            }
+            $requestPartitions = [];
+            foreach ($topicMeta->getPartitions() as $partitionMeta) {
+                $requestPartitions[] = new TopicPartition($topicName, $partitionMeta->getId());
+                $partitionMeta->getId();
+            }
+
+            return $this->consumer->getCommittedOffsets($requestPartitions, 1000);
+        }
+
+        return [];
+    }
+
+    public function getPartitionBounds(string $topicName, int $partitionId): array
+    {
+        $this->consumer->queryWatermarkOffsets($topicName, $partitionId, $low, $high, 1000);
+
+        return [$low, $high];
+    }
+
+    /**
+     * @throws RdKafkaException
+     */
+    public function setOffset(string $topicName, int $partitionId, int $offset): void
+    {
+        $this->consumer->commit([
+            new TopicPartition(
+                $topicName,
+                $partitionId,
+                $offset
+            )
+        ]);
+    }
 }

src/LaravelPhpRdKafkaConsumerServiceProvider.php

@@ -2,7 +2,9 @@
 
 namespace Ensi\LaravelPhpRdKafkaConsumer;
 
+use Ensi\LaravelPhpRdKafkaConsumer\Commands\KafkaCheckOffsetsCommand;
 use Ensi\LaravelPhpRdKafkaConsumer\Commands\KafkaConsumeCommand;
+use Ensi\LaravelPhpRdKafkaConsumer\Commands\KafkaSetOffset;
 use Illuminate\Support\ServiceProvider;
 
 class LaravelPhpRdKafkaConsumerServiceProvider extends ServiceProvider
@@ -26,6 +28,8 @@ public function boot()
 
             $this->commands([
                 KafkaConsumeCommand::class,
+                KafkaCheckOffsetsCommand::class,
+                KafkaSetOffset::class,
             ]);
         }
     }