add decoder encoder to kafka

Author: Iachimovschi Danu

src/Infrastructure/Command/UserConsumerCommand.php

@@ -16,6 +16,9 @@
 use Jobcloud\Kafka\Exception\KafkaConsumerConsumeException;
 use Jobcloud\Kafka\Exception\KafkaConsumerEndOfPartitionException;
 use Jobcloud\Kafka\Exception\KafkaConsumerTimeoutException;
+use Jobcloud\Kafka\Message\Decoder\AvroDecoder;
+use Jobcloud\Kafka\Message\Encoder\AvroEncoder;
+use Jobcloud\Kafka\Message\Registry\AvroSchemaRegistry;
 use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputArgument;
@@ -46,14 +49,15 @@ protected function execute(InputInterface $input, OutputInterface $output): int
     {
         $io = new SymfonyStyle($input, $output);
 
-
         $schemaRegistryClient = new CachedRegistry(
             new PromisingRegistry(
                 new Client(['base_uri' => 'schema-registry:8081'])
             ),
             new AvroObjectCacheAdapter()
         );
 
+        $registry = new AvroSchemaRegistry($schemaRegistryClient);
+
         $recordSerializer = new RecordSerializer(
             $schemaRegistryClient,
             [
@@ -62,19 +66,6 @@ protected function execute(InputInterface $input, OutputInterface $output): int
             ]
         );
 
-        $consumer = KafkaConsumerBuilder::create()
-            ->withAdditionalConfig(
-                [
-                    'enable.auto.commit' => false
-                ]
-            )
-            ->withAdditionalBroker('kafka:9092')
-            ->withConsumerGroup('testGroup')
-            ->withAdditionalSubscription('users')
-            ->build();
-
-        $consumer->subscribe();
-
         $schema = <<<'JSON'
         {
             "type": "record",
@@ -89,6 +80,22 @@ protected function execute(InputInterface $input, OutputInterface $output): int
 
         $avroSchema = AvroSchema::parse($schema);
 
+        $decoder = new AvroDecoder($registry, $recordSerializer);
+
+        $consumer = KafkaConsumerBuilder::create()
+            ->withAdditionalConfig(
+                [
+                    'enable.auto.commit' => false,
+                ]
+            )
+            ->withDecoder($decoder)
+            ->withAdditionalBroker('kafka:9092')
+            ->withConsumerGroup('testGroup')
+            ->withAdditionalSubscription('users')
+            ->build();
+
+        $consumer->subscribe();
+
         while (true) {
             try {
                 $message = $consumer->consume();

src/Infrastructure/Command/UserProducerCommand.php

@@ -11,7 +11,10 @@
 use FlixTech\SchemaRegistryApi\Registry\CachedRegistry;
 use FlixTech\SchemaRegistryApi\Registry\PromisingRegistry;
 use GuzzleHttp\Client;
+use Jobcloud\Kafka\Message\Decoder\AvroDecoder;
+use Jobcloud\Kafka\Message\Encoder\AvroEncoder;
 use Jobcloud\Kafka\Message\KafkaProducerMessage;
+use Jobcloud\Kafka\Message\Registry\AvroSchemaRegistry;
 use Jobcloud\Kafka\Producer\KafkaProducerBuilder;
 use League\Csv\Reader;
 use Symfony\Component\Console\Attribute\AsCommand;
@@ -37,6 +40,8 @@ protected function execute(InputInterface $input, OutputInterface $output): int
             new AvroObjectCacheAdapter()
         );
 
+        $registry = new AvroSchemaRegistry($schemaRegistryClient);
+
         $recordSerializer = new RecordSerializer(
             $schemaRegistryClient,
             [
@@ -60,7 +65,16 @@ protected function execute(InputInterface $input, OutputInterface $output): int
 
         $avroSchema = AvroSchema::parse($schema);
 
+        $encoder = new AvroEncoder($registry, $recordSerializer);
+
         $producer = KafkaProducerBuilder::create()
+            ->withAdditionalConfig(
+                [
+                    'compression.codec' => 'lz4',
+                    'auto.commit.interval.ms' => 500
+                ]
+            )
+            ->withEncoder($encoder)
             ->withAdditionalBroker('kafka:9092')
             ->build();