Throw NoAliveBrokerException when no brokers are available.

src/Broker.php

@@ -7,6 +7,7 @@
 use InvalidArgumentException;
 use longlang\phpkafka\Client\ClientInterface;
 use longlang\phpkafka\Consumer\ConsumerConfig;
+use longlang\phpkafka\Exception\NoAliveBrokerException;
 use longlang\phpkafka\Producer\ProducerConfig;
 use longlang\phpkafka\Protocol\ErrorCode;
 use longlang\phpkafka\Protocol\Metadata\MetadataRequest;
@@ -92,6 +93,11 @@ public function updateBrokers(): void
         foreach ($response->getBrokers() as $broker) {
             $brokers[$broker->getNodeId()] = $broker->getHost() . ':' . $broker->getPort();
         }
+
+        if (empty($brokers)) {
+            throw new NoAliveBrokerException('No brokers are available');
+        }
+
         $this->setBrokers($brokers);
     }
 

src/Exception/NoAliveBrokerException.php

@@ -0,0 +1,9 @@
+<?php
+
+declare(strict_types=1);
+
+namespace longlang\phpkafka\Exception;
+
+class NoAliveBrokerException extends \Exception
+{
+}

src/Producer/Producer.php

@@ -44,17 +44,17 @@ public function __construct(ProducerConfig $config)
         $this->partitioner = new $class();
     }
 
-    public function send(string $topic, ?string $value, ?string $key = null, array $headers = [], ?int $partitionIndex = null, ?int $brokerId = null): void
+    public function send(string $topic, ?string $value, ?string $key = null, array $headers = [], ?int $partitionIndex = null): void
     {
         $message = new ProduceMessage($topic, $value, $key, $headers, $partitionIndex);
         $messages = [$message];
-        $this->sendBatch($messages, $brokerId);
+        $this->sendBatch($messages);
     }
 
     /**
      * @param ProduceMessage[] $messages
      */
-    public function sendBatch(array $messages, ?int $brokerId = null): void
+    public function sendBatch(array $messages): void
     {
         $config = $this->config;
         $broker = $this->broker;