Allow to use custom options for producers

mateusjunges · Nov 5, 2021 · 38ca04c · 38ca04c
1 parent 45f7861
commit 38ca04c
Show file tree

Hide file tree

Showing 3 changed files with 56 additions and 1 deletion.
diff --git a/src/Config/Config.php b/src/Config/Config.php
@@ -83,7 +83,7 @@ public function getProducerOptions(): array
             'metadata.broker.list' => $this->broker,
         ];
 
-        return array_merge($config, $this->getSaslOptions());
+        return array_merge($config, $this->customOptions, $this->getSaslOptions());
     }
 
     private function getSaslOptions(): array

diff --git a/tests/Config/ConfigTest.php b/tests/Config/ConfigTest.php
@@ -102,4 +102,42 @@ public function testItReturnsProducerOptions()
             $config->getProducerOptions()
         );
     }
+
+    public function testItAcceptsCustomOptionsForProducersConfig()
+    {
+        $customOptions = [
+            'bootstrap.servers' => '[REMOTE_ADDRESS]',
+            'metadata.broker.list' => '[REMOTE_ADDRESS]',
+            'security.protocol' => 'SASL_SSL',
+            'sasl.mechanisms' => 'PLAIN',
+            'sasl.username' => '[API_KEY]',
+            'sasl.password' => '[API_KEY]',
+        ];
+
+        $config = new Config(
+            broker: 'broker',
+            topics: ['topic'],
+            securityProtocol: 'SASL_PLAINTEXT',
+            commit: 1,
+            groupId: 'group',
+            consumer: $this->createMock(Consumer::class),
+            dlq: null,
+            customOptions: $customOptions
+        );
+
+        $expectedOptions = [
+            'compression.codec' => 'snappy',
+            'bootstrap.servers' => '[REMOTE_ADDRESS]',
+            'metadata.broker.list' => '[REMOTE_ADDRESS]',
+            'security.protocol' => 'SASL_SSL',
+            'sasl.mechanisms' => 'PLAIN',
+            'sasl.username' => '[API_KEY]',
+            'sasl.password' => '[API_KEY]',
+        ];
+
+        $this->assertEquals(
+            $expectedOptions,
+            $config->getProducerOptions()
+        );
+    }
 }
diff --git a/tests/KafkaTest.php b/tests/KafkaTest.php
@@ -193,6 +193,23 @@ public function testICanDisableDebugUsingWithDebugDisabledMethod()
         $this->assertArrayNotHasKey('debug', $message->getHeaders());
     }
 
+    public function testICanUseCustomOptionsForProducerConfig()
+    {
+        $producer = Kafka::publishOn('localhost:9092', 'test-topic')
+            ->withConfigOptions($expectedOptions = [
+                'bootstrap.servers' => '[REMOTE_ADDRESS]',
+                'metadata.broker.list' => '[REMOTE_ADDRESS]',
+                'security.protocol' => 'SASL_SSL',
+                'sasl.mechanisms' => 'PLAIN',
+                'sasl.username' => '[API_KEY]',
+                'sasl.password' => '[API_KEY]',
+            ]);
+
+        $options = $this->getPropertyWithReflection('options', $producer);
+
+        $this->assertEquals($expectedOptions, $options);
+    }
+
     public function testCreateConsumerReturnsAConsumerBuilderInstance()
     {
         $consumer = Kafka::createConsumer('broker');