title: Apache Kafka LEADER_NOT_AVAILABLE
body: <p>I'm running into an issue with apache Kafka that I don't understand . I subscribe to a topic in my broker called "topic-received" . This is the code : </p>
  
  <pre><code>protected String readResponse(final String idMessage) {
      if (props != null) {
          kafkaClient = new KafkaConsumer&lt;&gt;(props);
          logger.debug("Subscribed to topic-received");
          kafkaClient.subscribe(Arrays.asList("topic-received"));
          logger.debug("Waiting for reading : topic-received");
          ConsumerRecords&lt;String, String&gt; records =    
                         kafkaClient.poll(kafkaConfig.getRead_timeout());
  
          if (records != null) {
              for (ConsumerRecord&lt;String, String&gt; record : records) {
                  logger.debug("Resultado devuelto : "+record.value());
                  return record.value();
              }
          }
      }
      return null;
  }
  </code></pre>
  
  <p>As this is happening, I send a message to "topic-received" from another point . The code is the following one : </p>
  
  <pre><code>private void sendMessageToKafkaBroker(String idTopic, String value) {
      Producer&lt;String, String&gt; producer = null;
      try {
          producer = new KafkaProducer&lt;String, String&gt;(mapProperties());
          ProducerRecord&lt;String, String&gt; producerRecord = new 
                     ProducerRecord&lt;String, String&gt;("topic-received", value);
          producer.send(producerRecord);
          logger.info("Sended value "+value+" to topic-received");
      } catch (ExceptionInInitializerError eix) {
          eix.printStackTrace();
      } catch (KafkaException ke) {
          ke.printStackTrace();
      } finally {
          if (producer != null) {
              producer.close();
          }
      }
  }
  </code></pre>
  
  <p>First time I try , with topic "topic-received", I get a warning like this </p>
  
  <pre><code>"WARN 13164 --- [nio-8085-exec-3] org.apache.kafka.clients.NetworkClient   :  
   Error while fetching metadata with correlation id 1 : {topic-  
   received=LEADER_NOT_AVAILABLE}"
  </code></pre>
  
  <p>But if I try again, to this topic "topic-received", works ok, and no warning is presented . Anyway, that's not useful for me, because I have to listen from a topic and send to a topic new each time ( referenced by an String identifier ex: .. 12Erw45-2345Saf-234DASDFasd )</p>
  
  <p>Looking for LEADER_NOT_AVAILABLE in google , some guys talk about adding to server.properties the next lines : </p>
  
  <pre><code>host.name=127.0.0.1
  advertised.port=9092
  advertised.host.name=127.0.0.1
  </code></pre>
  
  <p>But it's not working for me ( Don't know why ) . </p>
  
  <p>I have tried to create the topic before all this process with the following code:</p>
  
  <pre><code> private void createTopic(String idTopic) {
      String zookeeperConnect = "localhost:2181";
      ZkClient zkClient = new ZkClient(zookeeperConnect,10000,10000, 
      ZKStringSerializer$.MODULE$);
      ZkUtils zkUtils = new ZkUtils(zkClient, new 
      ZkConnection(zookeeperConnect),false);
      if(!AdminUtils.topicExists(zkUtils,idTopic)) {
          AdminUtils.createTopic(zkUtils, idTopic, 2, 1, new Properties(), 
      null);
          logger.debug("Created topic "+idTopic+" by super user");
      }
      else{
          logger.debug("topic "+idTopic+" already exists");
      }
    }
  </code></pre>
  
  <p>No error, but still, it stays listening till the timeout. </p>
  
  <p>I have reviewed the properties of the broker to check if there's any help, but I haven't found anything clear enough . The props that I have used for reading are : </p>
  
  <pre><code>    props = new Properties();
      props.put("bootstrap.servers", kafkaConfig.getBootstrap_servers());
      props.put("key.deserializer", kafkaConfig.getKey_deserializer());
      props.put("value.deserializer", kafkaConfig.getValue_deserializer());
      props.put("key.serializer", kafkaConfig.getKey_serializer());
      props.put("value.serializer", kafkaConfig.getValue_serializer());
      props.put("group.id",kafkaConfig.getGroupId());
  </code></pre>
  
  <p>and , for sending ...</p>
  
  <pre><code>   Properties props = new Properties();
      props.put("bootstrap.servers", kafkaConfig.getHost() + ":" + 
      kafkaConfig.getPort());
      props.put("group.id", kafkaConfig.getGroup_id());
      props.put("enable.auto.commit", kafkaConfig.getEnable_auto_commit());
      props.put("auto.commit.interval.ms", 
      kafkaConfig.getAuto_commit_interval_ms());
      props.put("session.timeout.ms", kafkaConfig.getSession_timeout_ms());
      props.put("key.deserializer", kafkaConfig.getKey_deserializer());
      props.put("value.deserializer", kafkaConfig.getValue_deserializer());
      props.put("key.serializer", kafkaConfig.getKey_serializer());
      props.put("value.serializer", kafkaConfig.getValue_serializer());
  </code></pre>
  
  <p>Any clue ? Why , the only way that I have to consume messages from the broker and from the topic, is repeating the request after an error ?</p>
  
  <p>Thanks in advance</p>