Multiple kafka broker connections

config/kafka.php

@@ -4,7 +4,19 @@
     /*
      | Your kafka brokers url.
      */
+    'default' => env('KAFKA_CONNECTION', 'main'),
     'brokers' => env('KAFKA_BROKERS', 'localhost:9092'),
+    'connections' => [
+        'main' => [
+            'brokers' => env('KAFKA_BROKERS', 'localhost:9092'),
+            'securityProtocol' => env('KAFKA_SECURITY_PROTOCOL', ''),
+            'sasl' => [
+                'username' => env('KAFKA_SASL_USERNAME', ''),
+                'password' => env('KAFKA_SASL_PASSWORD', ''),
+                'mechanisms' => env('KAFKA_SASL_MECHANISMS', ''),
+            ]
+        ],
+    ],
 
     /*
      | Kafka consumers belonging to the same consumer group share a group id.

src/Console/Commands/KafkaConsumer/Options.php

@@ -19,6 +19,7 @@ class Options
     private ?string $saslPassword;
     private ?string $saslMechanisms;
     private array $config;
+    private ?string $brokerConnection = null;
 
     #[Pure]
     public function __construct(array $options, array $config)
@@ -90,8 +91,16 @@ public function getSecurityProtocol(): ?string
         return $this->securityProtocol;
     }
 
-    public function getBroker()
+    public function getBroker(): string
     {
-        return $this->config['brokers'];
+        if ($this->brokerConnection) {
+            return $this->config['connections'][$this->brokerConnection]['brokers'];
+        }
+
+        if (!isset($this->config['default'])) {
+            return $this->config['brokers'];
+        }
+
+        return $this->config['connections'][$this->config['default']]['brokers'];
     }
 }

src/Console/Commands/KafkaConsumerCommand.php

@@ -18,7 +18,8 @@ class KafkaConsumerCommand extends Command
             {--commit=1} 
             {--dlq=? : The Dead Letter Queue} 
             {--maxMessage=? : The max number of messages that should be handled}
-            {--securityProtocol=?}';
+            {--securityProtocol=?}
+            {--brokerConnection= : The connection in kafka config to consume from}';
 
     protected $description = 'A Kafka Consumer for Laravel.';
 
@@ -28,14 +29,22 @@ public function __construct()
     {
         parent::__construct();
 
+        $defaultConnection = config('kafka.default');
+
+        $securityProtocol = config("kafka.connections.{$defaultConnection}.securityProtocol", config('kafka.securityProtocol'));
+
+        $sasl = config("kafka.connections.{$defaultConnection}.sasl", config('kafka.sasl', []));
+
         $this->config = [
             'brokers' => config('kafka.brokers'),
+            'connections' => config('kafka.connections'),
+            'default' => $defaultConnection,
             'groupId' => config('kafka.consumer_group_id'),
-            'securityProtocol' => config('kafka.securityProtocol'),
+            'securityProtocol' => $securityProtocol,
             'sasl' => [
-                'mechanisms' => config('kafka.sasl.mechanisms'),
-                'username' => config('kafka.sasl.username'),
-                'password' => config('kafka.sasl.password'),
+                'mechanisms' => $sasl['mechanism'] ?? null,
+                'username' => $sasl['username'] ?? null,
+                'password' => $sasl['password'] ?? null,
             ],
         ];
     }
@@ -53,6 +62,13 @@ public function handle()
 
             return;
         }
+
+        if ($this->option('brokerConnection')) {
+
+            $this->config['sasl'] = config("kafka.connections.{$this->option('brokerConnection')}.sasl");
+            $this->config['securityProtocol'] = config("kafka.connections.{$this->option('brokerConnection')}.securityProtocol");
+        }
+
         $options = new Options($this->options(), $this->config);
 
         $consumer = $options->getConsumer();

src/Kafka.php

@@ -19,9 +19,12 @@ class Kafka implements CanPublishMessagesToKafka, CanConsumeMessagesFromKafka
      */
     public function publishOn(string $topic, string $broker = null): CanProduceMessages
     {
+        $defaultBrokers = config('kafka.connections.' . config('kafka.default') . '.brokers') ??
+            config('kafka.brokers');
+
         return new ProducerBuilder(
             topic: $topic,
-            broker: $broker ?? config('kafka.brokers')
+            broker: $broker ?? $defaultBrokers
         );
     }
 
@@ -35,8 +38,11 @@ public function publishOn(string $topic, string $broker = null): CanProduceMessa
      */
     public function createConsumer(array $topics = [], string $groupId = null, string $brokers = null): ConsumerBuilder
     {
+        $defaultBrokers = config('kafka.connections.' . config('kafka.default') . '.brokers') ??
+            config('kafka.brokers');
+
         return ConsumerBuilder::create(
-            brokers: $brokers ?? config('kafka.brokers'),
+            brokers: $brokers ?? $defaultBrokers,
             topics: $topics,
             groupId: $groupId ?? config('kafka.consumer_group_id')
         );

src/Producers/ProducerBuilder.php

@@ -27,7 +27,10 @@ public function __construct(
         $message = app(KafkaProducerMessage::class);
         $this->message = $message->create($topic);
         $this->serializer = app(MessageSerializer::class);
-        $this->broker = $broker ?? config('kafka.brokers');
+
+        $defaultConnection = config('kafka.default');
+        $defaultBrokers = config('kafka.connections.' . $defaultConnection . '.brokers') ?? config('kafka.brokers');
+        $this->broker = $broker ?? $defaultBrokers;
     }
 
     /**

tests/Console/Consumers/OptionsTest.php

@@ -18,6 +18,7 @@ public function setUp(): void
             'brokers' => config('kafka.brokers'),
             'groupId' => config('kafka.group_id'),
             'securityProtocol' => config('kafka.securityProtocol'),
+            'connections' => config('kafka.connections'),
             'sasl' => [
                 'mechanisms' => config('kafka.sasl.mechanisms'),
                 'username' => config('kafka.sasl.username'),
@@ -70,4 +71,17 @@ public function testItInstantiatesUsingOnlyRequiredOptions()
         $this->assertEquals('plaintext', $options->getSecurityProtocol());
         $this->assertNull($options->getSasl());
     }
+
+    public function testItCanUseAnotherConnection()
+    {
+        $options = [
+            'topics' => 'test-topic,test-topic-1',
+            'consumer' => FakeHandler::class,
+            'brokerConnection' => 'test'
+        ];
+
+        $options = new Options($options, $this->config);
+
+        $this->assertEquals('localhost:9093', $options->getBroker());
+    }
 }

tests/KafkaTest.php

@@ -293,4 +293,33 @@ public function testSendMessageBatch()
 
         Kafka::publishOn('test')->withBodyKey('foo', 'bar')->sendBatch($messageBatch);
     }
+
+    public function testICanPublishMessagesOnTestConnection()
+    {
+        $mockedProducerTopic = m::mock(ProducerTopic::class)
+            ->shouldReceive('producev')->once()
+            ->andReturn(m::self())
+            ->getMock();
+
+        $mockedProducer = m::mock(Producer::class)
+            ->shouldReceive('newTopic')
+            ->andReturn($mockedProducerTopic)
+            ->shouldReceive('poll')
+            ->shouldReceive('flush')
+            ->andReturn(RD_KAFKA_RESP_ERR_NO_ERROR)
+            ->getMock();
+
+        $this->app->bind(Producer::class, function () use ($mockedProducer) {
+            return $mockedProducer;
+        });
+
+        $test = Kafka::publishOn('test-topic', config('kafka.broker_connections.test'))
+            ->withKafkaKey(Str::uuid()->toString())
+            ->withBodyKey('test', ['test'])
+            ->withHeaders(['custom' => 'header'])
+            ->withDebugEnabled()
+            ->send();
+
+        $this->assertTrue($test);
+    }
 }

tests/LaravelKafkaTestCase.php

@@ -27,6 +27,15 @@ public function setUp(): void
     public function getEnvironmentSetUp($app): void
     {
         $app['config']->set('kafka.brokers', 'localhost:9092');
+        $app['config']->set('kafka.default', 'main');
+        $app['config']->set('kafka.connections', [
+            'main' => [
+                'brokers' => 'localhost:9092'
+            ],
+            'test' => [
+                'brokers' => 'localhost:9093'
+            ],
+        ]);
         $app['config']->set('kafka.consumer_group_id', 'group');
         $app['config']->set('kafka.offset_reset', 'latest');
         $app['config']->set('kafka.auto_commit', true);