title: Kafka cluster loses messages after zookeeper restart
body: <p>I'm starting a cluster of kafka brokers using Docker (5 brokers for example, one broker per container). Kafka version 2.12-0.11.0.0, Zookeeper 3.4.10.</p>
  
  <p><strong>The scenario:</strong></p>
  
  <ul>
  <li>Starting 1st broker with config below</li>
  </ul>
  
  <p><em>zoo.cfg</em></p>
  
  <pre><code>tickTime=2000
  initLimit=10
  syncLimit=5
  
  dataDir=/opt/zookeeper/data
  
  clientPort=2181
  maxClientCnxns=10
  minSessionTimeout=4000
  maxSessionTimeout=1000000
  server.1=0.0.0.0:2888:3888
  </code></pre>
  
  <p><em>server.properties</em></p>
  
  <pre><code>broker.id=1
  listeners=PLAINTEXT://:9092
  advertised.listeners=PLAINTEXT://broker1_IP:broker1_PORT
  num.network.threads=3
  num.io.threads=8
  socket.send.buffer.bytes=102400
  socket.receive.buffer.bytes=102400
  socket.request.max.bytes=104857600
  log.dirs=/tmp/kafka-logs
  num.partitions=1
  num.recovery.threads.per.data.dir=1
  offsets.topic.replication.factor=1
  transaction.state.log.replication.factor=1
  transaction.state.log.min.isr=1
  log.retention.hours=168
  log.segment.bytes=1073741824
  log.retention.check.interval.ms=300000
  zookeeper.connect=127.0.0.1:2181
  zookeeper.session.timeout.ms=6000
  zookeeper.connection.timeout.ms=1000000
  group.initial.rebalance.delay.ms=0
  </code></pre>
  
  <p><em>producer.properties</em></p>
  
  <pre><code>bootstrap.servers=localhost:9092
  compression.type=none
  </code></pre>
  
  <p><em>consumer.properties</em></p>
  
  <pre><code>zookeeper.connect=127.0.0.1:2181
  zookeeper.session.timeout.ms=6000
  zookeeper.connection.timeout.ms=1000000
  group.id=test-consumer-group
  </code></pre>
  
  <ul>
  <li><p>Zookeeper is started in standalone mode, then starts kafka</p></li>
  <li><p>Creating topic</p></li>
  </ul>
  
  <p><code>/opt/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic my-test-topic1</code></p>
  
  <ul>
  <li>Sending message</li>
  </ul>
  
  <p><code>echo "test_kafka1" | /opt/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-test-topic1</code></p>
  
  <ul>
  <li>Checking message</li>
  </ul>
  
  <p><code>/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-test-topic1 --max-messages 1</code></p>
  
  <p><strong>Message is recieved</strong></p>
  
  <ul>
  <li>Describe the topic</li>
  </ul>
  
  <p><code>/opt/kafka/bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-test-topic1
  Topic:my-test-topic1    PartitionCount:1    ReplicationFactor:1 Configs:
      Topic: my-test-topic1   Partition: 0    Leader: 1   Replicas: 1 Isr: 1</code></p>
  
  <ul>
  <li>Starting rest 4 brokers</li>
  </ul>
  
  <p><em>zoo.cfg</em> on each broker from 1st to 5th (only 0.0.0.0:2888:3888 position differs)</p>
  
  <pre><code>tickTime=2000
  initLimit=10
  syncLimit=5
  
  dataDir=/opt/zookeeper/data
  
  clientPort=2181
  maxClientCnxns=10
  minSessionTimeout=4000
  maxSessionTimeout=1000000
  server.1=0.0.0.0:2888:3888
  server.2=broker2_IP:broker2_2888:broker2_3888
  server.3=broker3_IP:broker3_2888:broker3_3888
  server.4=broker4_IP:broker4_2888:broker4_3888
  server.5=broker5_IP:broker5_2888:broker5_3888
  </code></pre>
  
  <p><em>server.properties</em> on each broker from 1st to 5th (broker.id are unique, broker_IP:broker_PORT differs for ech broker)</p>
  
  <pre><code>broker.id=N
  listeners=PLAINTEXT://:9092
  advertised.listeners=PLAINTEXT://broker_IP:broker_PORT
  num.network.threads=3
  num.io.threads=8
  socket.send.buffer.bytes=102400
  socket.receive.buffer.bytes=102400
  socket.request.max.bytes=104857600
  log.dirs=/tmp/kafka-logs
  num.partitions=1
  num.recovery.threads.per.data.dir=1
  offsets.topic.replication.factor=1
  transaction.state.log.replication.factor=1
  transaction.state.log.min.isr=1
  log.retention.hours=168
  log.segment.bytes=1073741824
  log.retention.check.interval.ms=300000
  zookeeper.connect=127.0.0.1:2181
  zookeeper.session.timeout.ms=6000
  zookeeper.connection.timeout.ms=1000000
  group.initial.rebalance.delay.ms=0
  </code></pre>
  
  <p><em>producer.properties</em> on each broker from 1st to 5th</p>
  
  <pre><code>bootstrap.servers=localhost:9092
  compression.type=none
  </code></pre>
  
  <p><em>consumer.properties</em> on each broker from 1st to 5th</p>
  
  <pre><code>zookeeper.connect=127.0.0.1:2181
  zookeeper.session.timeout.ms=6000
  zookeeper.connection.timeout.ms=1000000
  group.id=test-consumer-group
  </code></pre>
  
  <ul>
  <li><p>Restarting zookeeper on each broker to take effect of zoo.cfg</p></li>
  <li><p>Zookeepers gather into cluster</p></li>
  <li><p>Topic moved to broker 5</p></li>
  </ul>
  
  <p><code>/opt/kafka/bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-test-topic1
  Topic:my-test-topic1    PartitionCount:1    ReplicationFactor:1 Configs:
      Topic: my-test-topic1   Partition: 0    Leader: 5   Replicas: 5 Isr: 5</code></p>
  
  <p>Is it normal behavior? Or should it stay on broker 1?</p>
  
  <ul>
  <li>Checking message on each broker</li>
  </ul>
  
  <p><code>/opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-test-topic1 --max-messages 1</code></p>
  
  <p><strong>Message is lost</strong> (message is not lost when topic stays on broker 1, so
     it is floating situation)</p>