Merge pull request #947 from dirk39/kafka_singleton_consumer

Kafka singleton consumer

Author: makasim

pkg/rdkafka/RdKafkaContext.php

@@ -45,13 +45,19 @@ class RdKafkaContext implements Context
      */
     private $kafkaConsumers;
 
+    /**
+     * @var RdKafkaConsumer[]
+     */
+    private $rdKafkaConsumers;
+
     /**
      * @param array $config
      */
     public function __construct(array $config)
     {
         $this->config = $config;
         $this->kafkaConsumers = [];
+        $this->rdKafkaConsumers = [];
 
         $this->setSerializer(new JsonSerializer());
     }
@@ -102,26 +108,33 @@ public function createConsumer(Destination $destination): Consumer
     {
         InvalidDestinationException::assertDestinationInstanceOf($destination, RdKafkaTopic::class);
 
-        $this->kafkaConsumers[] = $kafkaConsumer = new KafkaConsumer($this->getConf());
+        $queueName = $destination->getQueueName();
 
-        $consumer = new RdKafkaConsumer(
-            $kafkaConsumer,
-            $this,
-            $destination,
-            $this->getSerializer()
-        );
+        if (!isset($this->rdKafkaConsumers[$queueName])) {
+            $this->kafkaConsumers[] = $kafkaConsumer = new KafkaConsumer($this->getConf());
+
+            $consumer = new RdKafkaConsumer(
+                $kafkaConsumer,
+                $this,
+                $destination,
+                $this->getSerializer()
+            );
+
+            if (isset($this->config['commit_async'])) {
+                $consumer->setCommitAsync($this->config['commit_async']);
+            }
 
-        if (isset($this->config['commit_async'])) {
-            $consumer->setCommitAsync($this->config['commit_async']);
+            $this->rdKafkaConsumers[$queueName] = $consumer;
         }
 
-        return $consumer;
+        return $this->rdKafkaConsumers[$queueName];
     }
 
     public function close(): void
     {
         $kafkaConsumers = $this->kafkaConsumers;
         $this->kafkaConsumers = [];
+        $this->rdKafkaConsumers = [];
 
         foreach ($kafkaConsumers as $kafkaConsumer) {
             $kafkaConsumer->unsubscribe();

pkg/rdkafka/Tests/RdKafkaContextTest.php

@@ -69,4 +69,31 @@ public function testShouldInjectItsSerializerToConsumer()
 
         $this->assertSame($context->getSerializer(), $producer->getSerializer());
     }
+
+    public function testShouldNotCreateConsumerTwice()
+    {
+        $context = new RdKafkaContext(['global' => [
+            'group.id' => uniqid('', true),
+        ]]);
+        $queue = $context->createQueue('aQueue');
+
+        $consumer = $context->createConsumer($queue);
+        $consumer2 = $context->createConsumer($queue);
+
+        $this->assertSame($consumer, $consumer2);
+    }
+
+    public function testShouldCreateTwoConsumers()
+    {
+        $context = new RdKafkaContext(['global' => [
+            'group.id' => uniqid('', true),
+        ]]);
+        $queueA = $context->createQueue('aQueue');
+        $queueB = $context->createQueue('bQueue');
+
+        $consumer = $context->createConsumer($queueA);
+        $consumer2 = $context->createConsumer($queueB);
+
+        $this->assertNotSame($consumer, $consumer2);
+    }
 }