title: Kafka broker failover - Consumer not working when first node is down
body: <p>I am testing KAFKA(2.11_2.1.0) broker failover.</p>
  
  <p>I have 3 nodes:</p>
  
  <p>node0:    </p>
  
  <pre><code>broker.id=0 
  listeners=PLAINTEXT://localhost:9092
  advertised.listeners=PLAINTEXT://localhost:9092 
  </code></pre>
  
  <p>node1:</p>
  
  <pre><code>broker.id=1
  listeners=PLAINTEXT://localhost:9093
  advertised.listeners=PLAINTEXT://localhost:9093
  </code></pre>
  
  <p>node2:</p>
  
  <pre><code>broker.id=2
  listeners=PLAINTEXT://localhost:9094
  advertised.listeners=PLAINTEXT://localhost:9094
  </code></pre>
  
  <p>I created one topic with 3 partitions and 3 replicas.
  I created one producer with config:</p>
  
  <pre><code>properties.put("bootstrap.servers", "localhost:9092,localhost:9093,localhost:9094");
  properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, LongSerializer.class.getName());
  properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, MessageSerializer.class.getName());
  properties.put(ProducerConfig.ACKS_CONFIG, "1");
  properties.put(ProducerConfig.RETRIES_CONFIG, "3");
  properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");
  properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, MyPartitioner.class.getName());
  </code></pre>
  
  <p>I created 3 consumers for topic with config:</p>
  
  <pre><code>Properties properties = new Properties();
  properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092,localhost:9093,localhost:9094");
  properties.put(ConsumerConfig.GROUP_ID_CONFIG, appName);
  properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
  properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "3000");
  properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, LongDeserializer.class.getName());
  properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, MessageDeserializer.class.getName());
  properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, pulSize + "");
  properties.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "10000");
  properties.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, "3000");
  properties.put(ConsumerConfig.RECONNECT_BACKOFF_MS_CONFIG, "1000");
  </code></pre>
  
  <p><strong>Scenario 1</strong>:</p>
  
  <p>When <code>node0</code> stopped, consumers do not consume messages, but the producer produce messages and when start <code>node0</code>, everything is ok.</p>
  
  <p><strong>Scenario 2</strong>:</p>
  
  <p>When <code>node1</code> or <code>node2</code> stopped, consumers consume messages and the producer produce messages and everything is ok.</p>
  
  <p>Why failover not worked in scenario 1?</p>