test: add produce consume e2e test (#202)

Author: simPod

infection.json.dist

@@ -16,6 +16,6 @@
         },
         "text": "infection-log.txt"
     },
-    "minMsi": 40,
+    "minMsi": 35,
     "minCoveredMsi": 50
 }

tests/Clients/Consumer/KafkaBatchConsumerTest.php

@@ -22,7 +22,7 @@ public function testMaxBatchSize(): void
     {
         $testProducer = new TestProducer();
         for ($i = 0; $i < 100; $i++) {
-            $testProducer->run(self::PAYLOAD);
+            $testProducer->run(self::TOPIC, self::PAYLOAD);
         }
 
         $consumer = new KafkaConsumer($this->getConfig());

tests/Clients/Consumer/KafkaTest.php

@@ -0,0 +1,64 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Clients\Consumer;
+
+use Exception;
+use PHPUnit\Framework\TestCase;
+use SimPod\Kafka\Clients\Consumer\ConsumerConfig;
+use SimPod\Kafka\Clients\Consumer\KafkaConsumer;
+use SimPod\Kafka\Tests\Clients\Consumer\TestProducer;
+
+use function mt_rand;
+use function Safe\gethostname;
+
+use const RD_KAFKA_RESP_ERR__PARTITION_EOF;
+use const RD_KAFKA_RESP_ERR__TIMED_OUT;
+use const RD_KAFKA_RESP_ERR_NO_ERROR;
+
+final class KafkaTest extends TestCase
+{
+    public function testProduceConsume(): void
+    {
+        $topic   = 'produce-consume';
+        $headers = ['key' => 'value'];
+
+        $testProducer = new TestProducer();
+        $testProducer->run($topic, 'test', $headers);
+
+        unset($testProducer);
+
+        $consumer = new KafkaConsumer($this->getConfig());
+        $consumer->subscribe([$topic]);
+
+        while (true) {
+            $message = $consumer->consume(5000);
+            switch ($message->err) {
+                case RD_KAFKA_RESP_ERR_NO_ERROR:
+                    self::assertSame($message->headers, $headers);
+
+                    break 2;
+                // phpcs:ignore PSR2.ControlStructures.SwitchDeclaration.TerminatingComment
+                case RD_KAFKA_RESP_ERR__PARTITION_EOF:
+                    self::fail('No more messages; will wait for more');
+                // phpcs:ignore PSR2.ControlStructures.SwitchDeclaration.TerminatingComment
+                case RD_KAFKA_RESP_ERR__TIMED_OUT:
+                    self::fail('Timed out');
+                default:
+                    throw new Exception($message->errstr(), $message->err);
+            }
+        }
+    }
+
+    private function getConfig(): ConsumerConfig
+    {
+        $consumerConfig = new ConsumerConfig();
+        $consumerConfig->set(ConsumerConfig::BOOTSTRAP_SERVERS_CONFIG, '127.0.0.1:9092');
+        $consumerConfig->set(ConsumerConfig::CLIENT_ID_CONFIG, gethostname());
+        $consumerConfig->set(ConsumerConfig::GROUP_ID_CONFIG, mt_rand());
+        $consumerConfig->set(ConsumerConfig::AUTO_OFFSET_RESET_CONFIG, 'earliest');
+
+        return $consumerConfig;
+    }
+}

tests/Clients/Consumer/TestProducer.php

@@ -23,9 +23,15 @@ static function (KafkaProducer $producer): void {
         );
     }
 
-    public function run(string $payload): void
+    /** @param array<string, string>|null $headers */
+    public function run(string $topic, string $payload, array|null $headers = null): void
     {
-        $this->producer->produce(KafkaBatchConsumerTest::TOPIC, null, $payload);
+        $this->producer->produce(
+            $topic,
+            null,
+            $payload,
+            headers: $headers,
+        );
     }
 
     private function getConfig(): ProducerConfig