title: How to make Kafka broker failover to work regarding consumers?
body: <p>It seems very complicated to make a replicated broker work regarding consumers: it seems when stopping certain brokers, some consumers don't work anymore and, when the specific broker is up again, those consumers that didn't work receive all the "missing" messages.</p>
  
  <p>I am using a 2 brokers scenario. Created a replicated topic like this:</p>
  
  <pre><code>  $KAFKA_HOME/bin/kafka-topics.sh --create \
    --zookeeper localhost:2181 \
    --replication-factor 2 \
    --partitions 3 \
    --topic replicated_topic
  </code></pre>
  
  <p>The excerpt from the server config looks like this ( notice it is the same for both servers except port, broker id and log dir):</p>
  
  <pre><code># Licensed to the Apache Software Foundation (ASF) under one or more
  # contributor license agreements.  See the NOTICE file distributed with
  # this work for additional information regarding copyright ownership.
  # The ASF licenses this file to You under the Apache License, Version 2.0
  # (the "License"); you may not use this file except in compliance with
  # the License.  You may obtain a copy of the License at
  #
  #    http://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  # see kafka.server.KafkaConfig for additional details and defaults
  
  ############################# Server Basics #############################
  
  # The id of the broker. This must be set to a unique integer for each broker.
  broker.id=0
  ############################# Socket Server Settings #############################
  
  # The address the socket server listens on. It will get the value returned from 
  # java.net.InetAddress.getCanonicalHostName() if not configured.
  #   FORMAT:
  #     listeners = listener_name://host_name:port
  #   EXAMPLE:
  #     listeners = PLAINTEXT://your.host.name:9092
  listeners=PLAINTEXT://:9092
  
  # Hostname and port the broker will advertise to producers and consumers. If not set, 
  # it uses the value for "listeners" if configured.  Otherwise, it will use the value
  # returned from java.net.InetAddress.getCanonicalHostName().
  #advertised.listeners=PLAINTEXT://your.host.name:9092
  
  # Maps listener names to security protocols, the default is for them to be the same. See the config documentation for more details
  #listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL
  
  # The number of threads that the server uses for receiving requests from the network and sending responses to the network
  num.network.threads=3
  
  # The number of threads that the server uses for processing requests, which may include disk I/O
  num.io.threads=8
  
  # The send buffer (SO_SNDBUF) used by the socket server
  socket.send.buffer.bytes=102400
  
  # The receive buffer (SO_RCVBUF) used by the socket server
  socket.receive.buffer.bytes=102400
  
  # The maximum size of a request that the socket server will accept (protection against OOM)
  socket.request.max.bytes=104857600
  
  
  ############################# Log Basics #############################
  
  # A comma seperated list of directories under which to store log files
  log.dirs=/tmp/kafka-logs0
  
  # The default number of log partitions per topic. More partitions allow greater
  # parallelism for consumption, but this will also result in more files across
  # the brokers.
  num.partitions=1
  
  # The number of threads per data directory to be used for log recovery at startup and flushing at shutdown.
  # This value is recommended to be increased for installations with data dirs located in RAID array.
  num.recovery.threads.per.data.dir=1
  
  ############################# Internal Topic Settings  #############################
  # The replication factor for the group metadata internal topics "__consumer_offsets" and "__transaction_state"
  # For anything other than development testing, a value greater than 1 is recommended for to ensure availability such as 3.
  offsets.topic.replication.factor=2
  transaction.state.log.replication.factor=1
  transaction.state.log.min.isr=1
  
  ############################# Log Flush Policy #############################
  
  # Messages are immediately written to the filesystem but by default we only fsync() to sync
  # the OS cache lazily. The following configurations control the flush of data to disk.
  # There are a few important trade-offs here:
  #    1. Durability: Unflushed data may be lost if you are not using replication.
  #    2. Latency: Very large flush intervals may lead to latency spikes when the flush does occur as there will be a lot of data to flush.
  #    3. Throughput: The flush is generally the most expensive operation, and a small flush interval may lead to exceessive seeks.
  # The settings below allow one to configure the flush policy to flush data after a period of time or
  # every N messages (or both). This can be done globally and overridden on a per-topic basis.
  
  # The number of messages to accept before forcing a flush of data to disk
  #log.flush.interval.messages=10000
  
  # The maximum amount of time a message can sit in a log before we force a flush
  #log.flush.interval.ms=1000
  
  ############################# Log Retention Policy #############################
  
  # The following configurations control the disposal of log segments. The policy can
  # be set to delete segments after a period of time, or after a given size has accumulated.
  # A segment will be deleted whenever *either* of these criteria are met. Deletion always happens
  # from the end of the log.
  
  # The minimum age of a log file to be eligible for deletion due to age
  log.retention.hours=168
  
  # A size-based retention policy for logs. Segments are pruned from the log unless the remaining
  # segments drop below log.retention.bytes. Functions independently of log.retention.hours.
  #log.retention.bytes=1073741824
  
  # The maximum size of a log segment file. When this size is reached a new log segment will be created.
  log.segment.bytes=1073741824
  
  # The interval at which log segments are checked to see if they can be deleted according
  # to the retention policies
  log.retention.check.interval.ms=300000
  
  ############################# Zookeeper #############################
  
  # Zookeeper connection string (see zookeeper docs for details).
  # This is a comma separated host:port pairs, each corresponding to a zk
  # server. e.g. "127.0.0.1:3000,127.0.0.1:3001,127.0.0.1:3002".
  # You can also append an optional chroot string to the urls to specify the
  # root directory for all kafka znodes.
  zookeeper.connect=localhost:2181
  
  # Timeout in ms for connecting to zookeeper
  zookeeper.connection.timeout.ms=6000
  
  
  ############################# Group Coordinator Settings #############################
  
  # The following configuration specifies the time, in milliseconds, that the GroupCoordinator will delay the initial consumer rebalance.
  # The rebalance will be further delayed by the value of group.initial.rebalance.delay.ms as new members join the group, up to a maximum of max.poll.interval.ms.
  # The default value for this is 3 seconds.
  # We override this to 0 here as it makes for a better out-of-the-box experience for development and testing.
  # However, in production environments the default value of 3 seconds is more suitable as this will help to avoid unnecessary, and potentially expensive, rebalances during application startup.
  group.initial.rebalance.delay.ms=0
  </code></pre>
  
  <p>Let's decribe my topic using 2 brokers:</p>
  
  <pre><code>Topic:replicated_topic  PartitionCount:3    ReplicationFactor:2 Configs:
      Topic: replicated_topic Partition: 0    Leader: 1   Replicas: 1,0   Isr: 1,0
      Topic: replicated_topic Partition: 1    Leader: 0   Replicas: 0,1   Isr: 1,0
      Topic: replicated_topic Partition: 2    Leader: 1   Replicas: 1,0   Isr: 1,0
  </code></pre>
  
  <p>Let's see the code for the consumer:
  Consumer ( impl Callable )</p>
  
  <pre><code>@Override
  public Void call() throws Exception {
      final Properties props = new Properties();
      props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
              bootstrapServers);
      props.put(ConsumerConfig.GROUP_ID_CONFIG,
              groupId);
      props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
              IntegerDeserializer.class.getName());
      props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
              StringDeserializer.class.getName());
  
      final Consumer&lt;Integer, String&gt; consumer = new KafkaConsumer&lt;&gt;(props);
  
      consumer.subscribe(Collections.singletonList(topicName));
  
      ConsumerRecords&lt;Integer, String&gt; records = null;
      while (!Thread.currentThread().isInterrupted()) {
          records = consumer.poll(1000);
          if (records.isEmpty()) {
              continue;
          }
          records.forEach(rec -&gt; LOGGER.debug("{}@{} consumed from topic {}, partition {} pair ({},{})",
                  ConsumerCallable.class.getSimpleName(), hashCode(), rec.topic(), rec.partition(), rec.key(), rec.value()));
          consumer.commitAsync();
      }
  
      consumer.close();
      return null;
  }
  </code></pre>
  
  <p>Producer and main code:</p>
  
  <pre><code>private static final String TOPIC_NAME = "replicated_topic";
  private static final String BOOTSTRAP_SERVERS = "localhost:9092, localhost:9093";
  private static final Logger LOGGER = LoggerFactory.getLogger(Main.class);
  
  public static void main(String[] args) {
  
      ExecutorService executor = Executors.newCachedThreadPool();
      executor.submit(new ConsumerCallable(TOPIC_NAME, BOOTSTRAP_SERVERS, "group1"));
      executor.submit(new ConsumerCallable(TOPIC_NAME, BOOTSTRAP_SERVERS, "group2"));
      executor.submit(new ConsumerCallable(TOPIC_NAME, BOOTSTRAP_SERVERS, "group3"));
  
      try (Producer&lt;Integer, String&gt; producer = createProducer()) {
          Scanner scanner = new Scanner(System.in);
          String line = null;
          LOGGER.debug("Please enter 'k v' on the command line to send to Kafka or 'quit' to exit");
          while (scanner.hasNextLine()) {
              line = scanner.nextLine();
              if (line.trim().toLowerCase().equals("quit")) {
                  break;
              }
              String[] elements = line.split(" ");
              Integer key = Integer.parseInt(elements[0]);
              String value = elements[1];
              producer.send(new ProducerRecord&lt;&gt;(TOPIC_NAME, key, value));
              producer.flush();
          }
      }
      executor.shutdownNow();
  }
  
  private static Producer&lt;Integer, String&gt; createProducer() {
      Properties props = new Properties();
      props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
              BOOTSTRAP_SERVERS);
      props.put(ProducerConfig.CLIENT_ID_CONFIG, "KafkaExampleProducer");
      props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
              IntegerSerializer.class.getName());
      props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
              StringSerializer.class.getName());
      return new KafkaProducer&lt;&gt;(props);
  }
  </code></pre>
  
  <p>Now let's see the behaviour:</p>
  
  <ol>
  <li>All brokers are up:</li>
  </ol>
  
  <p>Output of kafka topic:</p>
  
  <pre><code>Topic:replicated_topic  PartitionCount:3    ReplicationFactor:2 Configs:
      Topic: replicated_topic Partition: 0    Leader: 1   Replicas: 1,0   Isr: 1,0
      Topic: replicated_topic Partition: 1    Leader: 0   Replicas: 0,1   Isr: 1,0
      Topic: replicated_topic Partition: 2    Leader: 1   Replicas: 1,0   Isr: 1,0
  </code></pre>
  
  <p>Output of program:</p>
  
  <pre><code>12:52:30.460 DEBUG Main - Please enter 'k v' on the command line to send to Kafka or 'quit' to exit
  1 u
  12:52:35.555 DEBUG ConsumerCallable - ConsumerCallable@1241910294 consumed from topic replicated_topic, partition 0 pair (1,u)
  12:52:35.559 DEBUG ConsumerCallable - ConsumerCallable@1361430455 consumed from topic replicated_topic, partition 0 pair (1,u)
  12:52:35.559 DEBUG ConsumerCallable - ConsumerCallable@186743616 consumed from topic replicated_topic, partition 0 pair (1,u)
  2 d
  12:52:38.096 DEBUG ConsumerCallable - ConsumerCallable@186743616 consumed from topic replicated_topic, partition 2 pair (2,d)
  12:52:38.098 DEBUG ConsumerCallable - ConsumerCallable@1361430455 consumed from topic replicated_topic, partition 2 pair (2,d)
  12:52:38.100 DEBUG ConsumerCallable - ConsumerCallable@1241910294 consumed from topic replicated_topic, partition 2 pair (2,d)
  </code></pre>
  
  <p>Since the consumers are in different groups all messages are broadcasted to them, everything is ok.</p>
  
  <p>2 Bring down broker 2:</p>
  
  <p>Describe topic:</p>
  
  <pre><code>Topic:replicated_topic  PartitionCount:3    ReplicationFactor:2 Configs:
      Topic: replicated_topic Partition: 0    Leader: 0   Replicas: 1,0   Isr: 0
      Topic: replicated_topic Partition: 1    Leader: 0   Replicas: 0,1   Isr: 0
      Topic: replicated_topic Partition: 2    Leader: 0   Replicas: 1,0   Isr: 0
  </code></pre>
  
  <p>Output of program:</p>
  
  <pre><code>3 t
  12:57:03.898 DEBUG ConsumerCallable - ConsumerCallable@186743616 consumed from topic replicated_topic, partition 1 pair (3,t)
  4 p
  12:57:06.058 DEBUG ConsumerCallable - ConsumerCallable@186743616 consumed from topic replicated_topic, partition 1 pair (4,p)
  </code></pre>
  
  <p>Now only 1 consumer receives data. Let's bring up broker 2 again:
  Now the other 2 consumers receive the missing data:</p>
  
  <pre><code>12:57:50.863 DEBUG ConsumerCallable - ConsumerCallable@1241910294 consumed from topic replicated_topic, partition 1 pair (3,t)
  12:57:50.863 DEBUG ConsumerCallable - ConsumerCallable@1241910294 consumed from topic replicated_topic, partition 1 pair (4,p)
  12:57:50.870 DEBUG ConsumerCallable - ConsumerCallable@1361430455 consumed from topic replicated_topic, partition 1 pair (3,t)
  12:57:50.870 DEBUG ConsumerCallable - ConsumerCallable@1361430455 consumed from topic replicated_topic, partition 1 pair (4,p)
  </code></pre>
  
  <ol start="3">
  <li>Bring down broker 1:</li>
  </ol>
  
  <p>Now only 2 consumers receive data:</p>
  
  <pre><code>5 c
  12:59:13.718 DEBUG ConsumerCallable - ConsumerCallable@1361430455 consumed from topic replicated_topic, partition 2 pair (5,c)
  12:59:13.737 DEBUG ConsumerCallable - ConsumerCallable@1241910294 consumed from topic replicated_topic, partition 2 pair (5,c)
  6 s
  12:59:16.437 DEBUG ConsumerCallable - ConsumerCallable@1361430455 consumed from topic replicated_topic, partition 2 pair (6,s)
  12:59:16.438 DEBUG ConsumerCallable - ConsumerCallable@1241910294 consumed from topic replicated_topic, partition 2 pair (6,s)
  </code></pre>
  
  <p>If I bring it on the other consumer wil also receive missing data.</p>
  
  <p>My point guys ( sorry for big write but I am trying to capture the context ), is how to make sure that no matter what broker I would stop, the consumers would work correctly? ( receive all messages normally )?</p>
  
  <p>PS: I tried setting the offsets.topic.replication.factor=2 or 3, but it didn't have any effect.</p>