Merge pull request #1264 from qkdreyer/patch-2

Allow rdkafka falsy keys

Author: makasim

pkg/rdkafka/RdKafkaProducer.php

@@ -37,9 +37,9 @@ public function send(Destination $destination, Message $message): void
         InvalidDestinationException::assertDestinationInstanceOf($destination, RdKafkaTopic::class);
         InvalidMessageException::assertMessageInstanceOf($message, RdKafkaMessage::class);
 
-        $partition = $this->getPartition($destination, $message);
+        $partition = $message->getPartition() ?? $destination->getPartition() ?? RD_KAFKA_PARTITION_UA;
         $payload = $this->serializer->toString($message);
-        $key = $message->getKey() ?: $destination->getKey() ?: null;
+        $key = $message->getKey() ?? $destination->getKey() ?? null;
 
         $topic = $this->producer->newTopic($destination->getTopicName(), $destination->getConf());
 
@@ -122,21 +122,4 @@ public function flush(int $timeout): void
             $this->producer->flush($timeout);
         }
     }
-
-    /**
-     * @param RdKafkaTopic   $destination
-     * @param RdKafkaMessage $message
-     */
-    private function getPartition(Destination $destination, Message $message): int
-    {
-        if (null !== $message->getPartition()) {
-            return $message->getPartition();
-        }
-
-        if (null !== $destination->getPartition()) {
-            return $destination->getPartition();
-        }
-
-        return \RD_KAFKA_PARTITION_UA;
-    }
 }

pkg/rdkafka/Tests/RdKafkaProducerTest.php

@@ -323,6 +323,66 @@ public function testShouldGetPartitionFromDestination(): void
         $producer->send($destination, $message);
     }
 
+    public function testShouldAllowFalsyKeyFromMessage(): void
+    {
+        $key = 0;
+
+        $kafkaTopic = $this->createKafkaTopicMock();
+        $kafkaTopic
+            ->expects($this->once())
+            ->method('producev')
+            ->with(
+                RD_KAFKA_PARTITION_UA,
+                0,
+                '',
+                $key
+            )
+        ;
+
+        $kafkaProducer = $this->createKafkaProducerMock();
+        $kafkaProducer
+            ->expects($this->once())
+            ->method('newTopic')
+            ->willReturn($kafkaTopic)
+        ;
+
+        $message = new RdKafkaMessage();
+        $message->setKey($key);
+
+        $producer = new RdKafkaProducer($kafkaProducer, $this->createSerializerMock());
+        $producer->send(new RdKafkaTopic(''), $message);
+    }
+
+    public function testShouldAllowFalsyKeyFromDestination(): void
+    {
+        $key = 0;
+
+        $kafkaTopic = $this->createKafkaTopicMock();
+        $kafkaTopic
+            ->expects($this->once())
+            ->method('producev')
+            ->with(
+                RD_KAFKA_PARTITION_UA,
+                0,
+                '',
+                $key
+            )
+        ;
+
+        $kafkaProducer = $this->createKafkaProducerMock();
+        $kafkaProducer
+            ->expects($this->once())
+            ->method('newTopic')
+            ->willReturn($kafkaTopic)
+        ;
+
+        $destination = new RdKafkaTopic('');
+        $destination->setKey($key);
+
+        $producer = new RdKafkaProducer($kafkaProducer, $this->createSerializerMock());
+        $producer->send($destination, new RdKafkaMessage());
+    }
+
     /**
      * @return \PHPUnit\Framework\MockObject\MockObject|ProducerTopic
      */