title: Kafka partition leader not updated after broker removed
body: <p>I have Kafka cluster managed by marathon/mesos which had 3 brokers version <code>0.10.2.1</code>. The docker images are based on <a href="https://github.com/wurstmeister/kafka-docker" rel="nofollow noreferrer">wurstmeister/kafka-docker</a>. The <code>broker.id=-1</code> which is assigned automatically and sequentially at start up and leaders are auto-rebalanced <code>auto.leader.rebalance.enable=true</code>. Clients are version <code>0.8.2.1</code>.</p>
  
  <p>Zookeeper configuration:
  </p>
  
  <pre class="lang-sh prettyprint-override"><code>➜ zkCli -server zookeeper.example.com:2181 ls /brokers/ids
  [1106, 1105, 1104]
  
  ➜ zkCli -server zookeeper.example.com:2181 get /brokers/ids/1104
  {"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},
  "endpoints":["PLAINTEXT://host1.mesos-slave.example.com:9092"],
  "jmx_port":9999,"host":"host1.mesos-slave.example.com",
  "timestamp":"1500987386409",
  "port":9092,"version":4}
  
  ➜ zkCli -server zookeeper.example.com:2181 get /brokers/ids/1105
  {"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},
  "endpoints":["PLAINTEXT://host2.mesos-slave.example.com:9092"],
  "jmx_port":9999,"host":"host2.mesos-slave.example.com",
  "timestamp":"1500987390304",
  "port":9092,"version":4}
  
  ➜ zkCli -server zookeeper.example.com:2181 get /brokers/ids/1106
  {"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},
  "endpoints":["PLAINTEXT://host3.mesos-slave.example.com:9092"],
  "jmx_port":9999,"host":"host3.mesos-slave.example.com",
  "timestamp":"1500987390447","port":9092,"version":4}
  
  ➜ bin/kafka-topics.sh --zookeeper zookeeper.example.com:2181 --create --topic test-topic --partitions 2 --replication-factor 2
  Created topic "test-topic".
  
  ➜ bin/kafka-topics.sh --zookeeper zookeeper.example.com:2181 --describe --topic test-topic
  Topic:test-topic    PartitionCount:2        ReplicationFactor:2     Configs:
          Topic: test-topic  Partition: 0    Leader: 1106    Replicas: 1106,1104     Isr: 1106
          Topic: test-topic  Partition: 1    Leader: 1105    Replicas: 1104,1105     Isr: 1105
  </code></pre>
  
  <p>Consumers can consume what producers are outputting. </p>
  
  <pre class="lang-sh prettyprint-override"><code>➜ /opt/kafka_2.10-0.8.2.1 bin/kafka-console-producer.sh --broker-list 10.0.1.3:9092,10.0.1.1:9092 --topic test-topic
  [2017-07-25 12:57:17,760] WARN Property topic is not valid (kafka.utils.VerifiableProperties)
  hello 1
  hello 2
  hello 3
  ...
  
  ➜ /opt/kafka_2.10-0.8.2.1 bin/kafka-console-consumer.sh --zookeeper zookeeper.example.com:2181 --topic test-topic --from-beginning
  hello 1
  hello 2
  hello 3
  ...
  </code></pre>
  
  <p>Then broker 1104 and 1105 (host1 and host2) go out and another one is coming online, 1107 (host 1), manually using marathon interface</p>
  
  <pre class="lang-sh prettyprint-override"><code>➜ zkCli -server zookeeper.example.com:2181 ls /brokers/ids
  [1107, 1106]
  
  ➜ zkCli -server zookeeper.example.com:2181 get /brokers/ids/1107
  {"listener_security_protocol_map":{"PLAINTEXT":"PLAINTEXT"},
  "endpoints":["PLAINTEXT://host1.mesos-slave.example.com:9092"],
  "jmx_port":9999,"host":"host1.mesos-slave.example.com",
  "timestamp":"1500991298225","port":9092,"version":4}
  </code></pre>
  
  <p>Consumer still gets messages from the producer but the topics description looks out of date:</p>
  
  <pre class="lang-sh prettyprint-override"><code>Topic:test-topic    PartitionCount:2        ReplicationFactor:2     Configs:
          Topic: test-topic  Partition: 0    Leader: 1106    Replicas: 1106,1104     Isr: 1106
          Topic: test-topic  Partition: 1    Leader: 1105    Replicas: 1104,1105     Isr: 1105
  </code></pre>
  
  <p>I tried rebalancing the <code>kafka-preferred-replica-election.sh</code>, <code>kafka-reassign-partitions.sh</code>.</p>
  
  <pre class="lang-sh prettyprint-override"><code>➜ $cat all_partitions.json
  {
    "version":1,
    "partitions":[
      {"topic":"test-topic","partition":0,"replicas":[1106,1107]},
      {"topic":"test-topic","partition":1,"replicas":[1107,1106]}
    ]
  }
  
  ➜ bin/kafka-reassign-partitions.sh --zookeeper zookeeper.example.com:2181 --reassignment-json-file all_partitions.json --execute
  
  ➜ bin/kafka-reassign-partitions.sh --zookeeper zookeeper.example.com:2181 --reassignment-json-file all_partitions.json --verify
  
  Status of partition reassignment:
  Reassignment of partition [test-topic,0] completed successfully
  Reassignment of partition [test-topic,1] is still in progress
  
  ➜ $cat all_leaders.json
  {
    "partitions":[
      {"topic": "test-topic", "partition": 0},
      {"topic": "test-topic", "partition": 1}
    ]
  }
  
  ➜ bin/kafka-preferred-replica-election.sh --zookeeper zookeeper.example.com:2181 --path-to-json-file all_leaders.json
  Created preferred replica election path with {"version":1,"partitions":[{"topic":"test-topic","partition":0},{"topic":"test-topic","partition":1}]}
  Successfully started preferred replica election for partitions Set([test-topic,0], [test-topic,1])
  </code></pre>
  
  <p>The partition leader for partitions 1 is still 1105 which doesn't make any sense:</p>
  
  <pre class="lang-sh prettyprint-override"><code>➜ bin/kafka-topics.sh --zookeeper zookeeper.example.com:2181 --describe --topic test-topic
  
  Topic:test-topic    PartitionCount:2        ReplicationFactor:2     Configs:
          Topic: test-topic   Partition: 0    Leader: 1106    Replicas: 1106,1107     Isr: 1106,1107
          Topic: test-topic   Partition: 1    Leader: 1105    Replicas: 1107,1106,1104,1105   Isr: 1105
  </code></pre>
  
  <p>Why partition 1 thinks that leader is still 1105 although host2 is not alive?</p>