title: Kafka broker may not be available exception
body: <p>I faced a strange issue with my Kafka producer. I use kafka-0.11 server/client version.
  I have one zookeper and one kafka broker node. Also, I created 'events' topic with 3 partitions:</p>
  
  <pre><code>Topic:events    PartitionCount:3        ReplicationFactor:1     Configs:
          Topic: events   Partition: 0    Leader: 0       Replicas: 0     Isr: 0
          Topic: events   Partition: 1    Leader: 0       Replicas: 0     Isr: 0
          Topic: events   Partition: 2    Leader: 0       Replicas: 0     Isr: 0
  </code></pre>
  
  <p>In my java code I create producer with the following properties:</p>
  
  <pre><code>Properties props = new Properties();
  props.put(BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
  props.put(MAX_BLOCK_MS_CONFIG, 30000);
  props.put(KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  props.put(VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
  props.put(PARTITIONER_CLASS_CONFIG, KafkaCustomPartitioner.class);
  this.producer = new KafkaProducer&lt;&gt;(props);
  </code></pre>
  
  <p>Also, I'have added a callback to Producer#send() method that adds failed message to the queue that is iterated by the other "re-sending" thread in a loop:</p>
  
  <pre><code>this.producer.send(producerRecord, new ProducerCallback(producerRecord.value(), topic));
  
  private class ProducerCallback implements Callback {
    private final String message;
    private final String topic;
  
    public ProducerCallback(String message, String topic) {
      this.message = message;
      this.topic = topic;
    }
  
    @Override
    public void onCompletion(RecordMetadata metadata, Exception ex) {
      if (ex != null) {
          logger.error("Kafka producer error. Topic: " + topic +
                  ".Message will be added into failed messages queue.", ex);
          failedMessagesQueue.enqueue(SerializationUtils.serialize(new FailedMessage(topic, message)));
      }
    }
  }
  
  private class ResenderThread extends Thread {
      private volatile boolean running = true;
  
      public void stopGracefully() {
          running = false;
      }
  
      @Override
      public void run() {
          while (running) {
              try {
                  byte[] val = failedMessagesQueue.peek();
                  if (val != null) {
                      FailedMessage failedMessage = SerializationUtils.deserialize(val);
                      ProducerRecord&lt;String, String&gt; record;
                      if (topic.equals(failedMessage.getTopic())) {
                          String messageKey = generateMessageKey(failedMessage.getMessage());
                          record = createProducerRecordWithKey(failedMessage.getMessage(), messageKey, failedMessage.getTopic());
                      } else {
                          record = new ProducerRecord&lt;&gt;(failedMessage.getTopic(), failedMessage.getMessage());
                      }
                      try {
                          this.producer.send(record).get();
                          failedMessagesQueue.dequeue();
                      } catch (Exception e) {
                          logger.debug("Kafka message resending attempt was failed. Topic " + failedMessage.getTopic() +
                                  " Partition. " + record.partition() + ". " + e.getMessage());
                      }
                  }
  
                  Thread.sleep(200);
              } catch (Exception e) {
                  logger.error("Error resending an event", e);
                  break;
              }
          }
      }
  }
  </code></pre>
  
  <p>Everything works fine until I decided to test Kafka broker kill/re-start scenario:</p>
  
  <p>I've killed my Kafka broker node, and sent a 5 messages using my Kafka producer. The following message were logged by my producer app:</p>
  
  <pre><code>....the application works fine....
  // kafka broker was killed
  2017-11-10 09:20:44,594 WARN [org.apache.kafka.clients.NetworkClient] - &lt;Connection to node 0 could not be established. Broker may not be available.&gt;
  2017-11-10 09:20:44,646 WARN [org.apache.kafka.clients.NetworkClient] - &lt;Connection to node 0 could not be established. Broker may not be available.&gt;
  2017-11-10 09:20:44,700 WARN [org.apache.kafka.clients.NetworkClient] - &lt;Connection to node 0 could not be established. Broker may not be available.&gt;
  2017-11-10 09:20:44,759 WARN [org.apache.kafka.clients.NetworkClient] - &lt;Connection to node 0 could not be established. Broker may not be available.&gt;
  2017-11-10 09:20:44,802 WARN [org.apache.kafka.clients.NetworkClient] - &lt;Connection to node 0 could not be established. Broker may not be available.&gt;
  // sent 5 message using producer. message were put to the failedMessagesQueue and "re-sender" thread started resending 
  2017-11-10 09:20:44,905 ERROR [com.inq.kafka.KafkaETLService] - &lt;Kafka producer error. Topic: events.Message will be added into failed messages queue.&gt;
  ....
  2017-11-10 09:20:45,070 WARN [org.apache.kafka.clients.NetworkClient] - &lt;Connection to node 0 could not be established. Broker may not be available.&gt;
  2017-11-10 09:20:45,129 WARN [org.apache.kafka.clients.NetworkClient] - &lt;Connection to node 0 could not be established. Broker may not be available.&gt;
  2017-11-10 09:20:45,170 WARN [org.apache.kafka.clients.NetworkClient] - &lt;Connection to node 0 could not be established. Broker may not be available.&gt;
  2017-11-10 09:20:45,217 WARN [org.apache.kafka.clients.NetworkClient] - &lt;Connection to node 0 could not be established. Broker may not be available.&gt;
  
  // kafka broker was restarted, some strange errors were logged
  2017-11-10 09:20:51,103 WARN [org.apache.kafka.clients.NetworkClient] - &lt;Error while fetching metadata with correlation id 29 : {events=INVALID_REPLICATION_FACTOR}&gt;
  2017-11-10 09:20:51,205 WARN [org.apache.kafka.clients.NetworkClient] - &lt;Error while fetching metadata with correlation id 31 : {events=INVALID_REPLICATION_FACTOR}&gt;
  2017-11-10 09:20:51,308 WARN [org.apache.kafka.clients.NetworkClient] - &lt;Error while fetching metadata with correlation id 32 : {events=INVALID_REPLICATION_FACTOR}&gt;
  2017-11-10 09:20:51,114 WARN [org.apache.kafka.clients.producer.internals.Sender] - &lt;Received unknown topic or partition error in produce request on partition events-0. The topic/partition may not exist or the user may not have Describe access to it&gt;
  2017-11-10 09:20:51,114 ERROR [com.inq.kafka.KafkaETLService] - &lt;Kafka message resending attempt was failed. Topic events. org.apache.kafka.common.errors.UnknownTopicOrPartitionException: This server does not host this topic-partition.&gt;
  2017-11-10 09:20:52,485 WARN [org.apache.kafka.clients.NetworkClient] - &lt;Error while fetching metadata with correlation id 33 : {events=INVALID_REPLICATION_FACTOR}&gt;
  // messages were succesfully re-sent and received by consumer..
  </code></pre>
  
  <p>How can I get rid of these logs (that logs every 100ms when Kafka broker is down):</p>
  
  <pre><code>[org.apache.kafka.clients.NetworkClient] - &lt;Connection to node 0 could not be established. Broker may not be available.&gt;
  </code></pre>
  
  <p>Why do I receive the following errors after Kafka broker startup(I didn't change any server props, and didn't alter the topic). It seems to me that these errors are result of some syncrhonization process between zookeeper and kafka during broker startup, because after some time procuder succesfully resent my messages. Am I wrong?:</p>
  
  <pre><code>[org.apache.kafka.clients.NetworkClient] - &lt;Error while fetching metadata with correlation id 29 : {events=INVALID_REPLICATION_FACTOR}&gt;
  Received unknown topic or partition error in produce request on partition events-0. The topic/partition may not exist or the user may not have Describe access to it. 
  </code></pre>