Auto create topics producer

src/Config.php

@@ -24,6 +24,7 @@
  * @method int getMetadataRequestTimeoutMs()
  * @method int getMetadataRefreshIntervalMs()
  * @method int getMetadataMaxAgeMs()
+ * @method bool getAutoCreateTopicsEnable()
  * @method string getSecurityProtocol()
  * @method bool getSslEnable()
  * @method void setSslEnable(bool $sslEnable)
@@ -84,6 +85,7 @@ abstract class Config
         'metadataRequestTimeoutMs'  => 60000,
         'metadataRefreshIntervalMs' => 300000,
         'metadataMaxAgeMs'          => -1,
+        'autoCreateTopicsEnable'    => true,
         'securityProtocol'          => self::SECURITY_PROTOCOL_PLAINTEXT,
         'sslEnable'                 => false, // this config item will override, don't config it.
         'sslLocalCert'              => '',
@@ -241,6 +243,11 @@ public function setMetadataMaxAgeMs(int $metadataMaxAgeMs): void
         static::$options['metadataMaxAgeMs'] = $metadataMaxAgeMs;
     }
 
+    public function setAutoCreateTopicsEnable(bool $flag = true): void
+    {
+        static::$options['autoCreateTopicsEnable'] = $flag;
+    }
+
     /**
      * @throws Exception\Config
      */

src/Producer/RecordValidator.php

@@ -4,6 +4,7 @@
 namespace Kafka\Producer;
 
 use Kafka\Exception;
+use Kafka\ProducerConfig;
 use function is_string;
 use function trim;
 
@@ -30,7 +31,10 @@ public function validate(array $record, array $topicList): void
             throw Exception\InvalidRecordInSet::missingTopic();
         }
 
-        if (! isset($topicList[$record['topic']])) {
+        /** @var ProducerConfig $config */
+        $config = ProducerConfig::getInstance();
+
+        if (! isset($topicList[$record['topic']]) && ! $config->getAutoCreateTopicsEnable()) {
             throw Exception\InvalidRecordInSet::nonExististingTopic($record['topic']);
         }
 

tests/Base/Producer/RecordValidatorTest.php

@@ -5,6 +5,7 @@
 
 use Kafka\Exception\InvalidRecordInSet;
 use Kafka\Producer\RecordValidator;
+use Kafka\ProducerConfig;
 use PHPUnit\Framework\TestCase;
 
 final class RecordValidatorTest extends TestCase
@@ -16,6 +17,10 @@ public function setUp(): void
     {
         $this->recordValidator = new RecordValidator();
 
+        /** @var ProducerConfig $config */
+        $config = ProducerConfig::getInstance();
+        $config->setAutoCreateTopicsEnable(false);
+
         parent::setUp();
     }
 
@@ -61,4 +66,17 @@ public function invalidRecordThrowsExceptionDataProvider(): array
             ],
         ];
     }
+
+    /**
+     * @doesNotPerformAssertions
+     * @phpcsSuppress SlevomatCodingStandard.TypeHints.TypeHintDeclaration.UselessDocComment
+     */
+    public function testValidateNonExistingTopicWhenAutoCreateTopicsEnabled(): void
+    {
+        /** @var ProducerConfig $config */
+        $config = ProducerConfig::getInstance();
+        $config->setAutoCreateTopicsEnable(true);
+
+        $this->recordValidator->validate(['topic' => 'test', 'value' => 'a value'], []);
+    }
 }

tests/Functional/Ticket/GH181Test.php

@@ -22,9 +22,11 @@ public function setUp(): void
             );
         }
 
+        /** @var ProducerConfig $config */
         $config = ProducerConfig::getInstance();
         $config->setMetadataBrokerList($brokers);
         $config->setBrokerVersion($version);
+        $config->setAutoCreateTopicsEnable(false);
 
         parent::setUp();
     }