add support auth

Author: mxp100

README.md

@@ -13,6 +13,13 @@ composer require laravel-tool/kafka-queue
 'kafka' => [
     'driver' => 'kafka',
     'broker_list' => env('KAFKA_BROKER_LIST', 'kafka:9092'),
+    'auth' => [
+        'enable' => env('KAFKA_AUTH_ENABLE', false),
+        'mechanism' => env('KAFKA_AUTH_MECHANISM', 'PLAIN'),
+        'username' => env('KAFKA_AUTH_USERNAME'),
+        'password' => env('KAFKA_AUTH_PASSWORD'),
+        'ssl_ca_location' => env('KAFKA_AUTH_SSL_CA_LOCATION'),
+    ],
     'queue' => env('KAFKA_QUEUE', 'default'),
     'heartbeat_ms' => env('KAFKA_HEARTBEAT', 5000),
     'group_name' => env('KAFKA_QUEUE_GROUP', 'default'),

src/Kafka/AuthConfigTrait.php

@@ -0,0 +1,20 @@
+<?php
+
+namespace LaravelTool\KafkaQueue\Kafka;
+
+use RdKafka\Conf as KafkaConfig;
+
+trait AuthConfigTrait
+{
+    public function authConfig(KafkaConfig &$kafkaConfig, array $config): void
+    {
+        if (!empty($config['auth']) && $config['auth']['enable']) {
+            $kafkaConfig->set('sasl.mechanisms', $config['auth']['mechanism']);
+            $kafkaConfig->set('sasl.username', $config['auth']['username']);
+            $kafkaConfig->set('sasl.password', $config['auth']['password']);
+            if ($config['auth']['ssl_ca_location']) {
+                $kafkaConfig->set('ssl.ca.location', $config['auth']['ssl_ca_location']);
+            }
+        }
+    }
+}

src/Kafka/Consumer.php

@@ -12,6 +12,8 @@
 
 class Consumer
 {
+    use AuthConfigTrait;
+
     private KafkaConsumer $consumer;
     private Metadata $metadata;
     private array $topics;
@@ -24,7 +26,9 @@ public function __construct(
     ) {
         $this->consumer = new KafkaConsumer($this->generateConfig($this->config));
 
-        $this->metadata();
+        if (!$this->config['queue_disable_length']) {
+            $this->metadata();
+        }
     }
 
     public function consume(string $topic): ?Message
@@ -54,6 +58,9 @@ public function commit(): void
 
     public function size(string $topicName): int
     {
+        if ($this->config['queue_disable_length']) {
+            return 0;
+        }
 
         try {
             if ($topic = $this->getTopic($topicName)) {
@@ -115,6 +122,8 @@ private function generateConfig(array $config): KafkaConfig
         $kafkaConfig->set('heartbeat.interval.ms', $config['heartbeat_ms']);
         $kafkaConfig->set('auto.offset.reset', 'earliest');
 
+        $this->authConfig($kafkaConfig, $config);
+
         return $kafkaConfig;
     }
 

src/Kafka/Producer.php

@@ -9,6 +9,8 @@
 
 class Producer
 {
+    use AuthConfigTrait;
+
     private KafkaProducer $producer;
 
     public function __construct(
@@ -49,6 +51,8 @@ private function generateConfig(array $config): KafkaConfig
         $kafkaConfig = new KafkaConfig();
         $kafkaConfig->set('metadata.broker.list', $config['broker_list']);
 
+        $this->authConfig($kafkaConfig, $config);
+
         return $kafkaConfig;
     }
 }