A broker going down or coming up causes a never-ending loop of rebalances

[dantswain]

We have been having hardware issues with one of our brokers. Whenever it goes down, our consumers (using ConsumerGroup) end up in an endless cycle of rebalancing the consumer group. The only way to resolve this is to stop all of the consumers in the group for around 60 seconds so that the broker recognizes that the consumers are all "dead", then restart the consumers. The same thing happens when the down broker comes back online and gets elected by the cluster as a leader for some partitions.

This may be related to #246 or may be the same underlying issue - I am not 100% certain as we've been in firefighting mode.

This may also be related to the following: https://issues.apache.org/jira/browse/KAFKA-2985

[dantswain]

Status update. I'm unable to reproduce this yet with a local cluster, but it's a big issue for us in production so I am heavily invested in fixing it.

[dantswain]

cjpoll in #kafkaex slack appears to be noticing a similar problem.

We're experiencing an issue from within the Consumer Group that seems to be causing quite a bit of churn for one partition. We don't seem to be able to progress past a particular point in the stream and run into this:

** (MatchError) no match of right hand side value: %KafkaEx.Protocol.LeaveGroup.Response{error_code: :unknown_member_id}
    (kafka_ex) lib/kafka_ex/consumer_group/manager.ex:338: KafkaEx.ConsumerGroup.Manager.leave/1
    (kafka_ex) lib/kafka_ex/consumer_group/manager.ex:167: KafkaEx.ConsumerGroup.Manager.terminate/2
    (stdlib) gen_server.erl:629: :gen_server.try_terminate/3
    (stdlib) gen_server.erl:795: :gen_server.terminate/7
    (stdlib) proc_lib.erl:247: :proc_lib.init_p_do_apply/3
Last message: :heartbeat
  neighbours:
=CRASH REPORT==== 16-Nov-2017::22:04:55 ===
  crasher:
    initial call: Elixir.KafkaEx.ConsumerGroup.Manager:init/1
    pid: <0.4153.0>
    registered_name: []
    exception exit: {{badmatch,
                         #{'__struct__' =>
                               'Elixir.KafkaEx.Protocol.LeaveGroup.Response',
                           error_code => unknown_member_id}},
                     [{'Elixir.KafkaEx.ConsumerGroup.Manager',leave,1,
                          [{file,"lib/kafka_ex/consumer_group/manager.ex"},
                           {line,338}]},
                      {'Elixir.KafkaEx.ConsumerGroup.Manager',terminate,2,
                          [{file,"lib/kafka_ex/consumer_group/manager.ex"},
                           {line,167}]},
                      {gen_server,try_terminate,3,
                          [{file,"gen_server.erl"},{line,629}]},
                      {gen_server,terminate,7,
                          [{file,"gen_server.erl"},{line,795}]},
                      {proc_lib,init_p_do_apply,3,
                          [{file,"proc_lib.erl"},{line,247}]}]}
      in function  gen_server:terminate/7 (gen_server.erl, line 800)
    ancestors: [<0.426.0>,'Elixir.<OurApp>.Supervisor',<0.411.0>]
    messages: []
    links: [<0.426.0>]
    dictionary: []
    trap_exit: true
    status: running
    heap_size: 1598
    stack_size: 27
    reductions: 910
22:04:55.117 [error] GenServer #PID<0.4153.0> terminating

We've frequently seen this error. Would this cause the consumer group supervisor to crash?

=CRASH REPORT==== 16-Nov-2017::22:12:31 ===
  neighbours:
    initial call: Elixir.KafkaEx.ConsumerGroup.Manager:init/1
    pid: <0.3538.0>
    registered_name: []
    exception exit: {{case_clause,{error,{already_started,<0.3540.0>}}},
                     [{'Elixir.KafkaEx.ConsumerGroup',start_consumer,5,
                          [{file,"lib/kafka_ex/consumer_group.ex"},
                           {line,261}]},
                      {'Elixir.KafkaEx.ConsumerGroup.Manager',start_consumer,
                          2,
                          [{file,"lib/kafka_ex/consumer_group/manager.ex"},
                           {line,390}]},
22:12:31.563 [error] GenServer #PID<0.3538.0> terminating
                          [{file,"lib/kafka_ex/consumer_group/manager.ex"},
                           {line,276}]},
                      {'Elixir.KafkaEx.ConsumerGroup.Manager',handle_info,2,
                          [{file,"lib/kafka_ex/consumer_group/manager.ex"},
                           {line,149}]},
                      {gen_server,try_dispatch,4,
                          [{file,"gen_server.erl"},{line,601}]},
                      {gen_server,handle_msg,5,
                          [{file,"gen_server.erl"},{line,667}]},
                      {proc_lib,init_p_do_apply,3,
                          [{file,"proc_lib.erl"},{line,247}]}]}
      in function  gen_server:terminate/7 (gen_server.erl, line 812)
    ancestors: [<0.426.0>,'Elixir.BlackMamba.Supervisor',<0.411.0>]
    messages: []
    links: [<0.426.0>]
    dictionary: []
    trap_exit: true
    status: running
    heap_size: 1598
    stack_size: 27
    reductions: 682
    (kafka_ex) lib/kafka_ex/consumer_group/manager.ex:149: KafkaEx.ConsumerGroup.Manager.handle_info/2

[CJPoll]

It does appear to be related to either network connectivity issues or a flapping instance:

22:29:31.567 [error] Receiving data from broker "10.1.12.11":9092 failed with :timeout
22:29:31.537 [error] Receiving data from broker "10.1.12.11":9092 failed with :timeout

[CJPoll]

A coworker of mine was able to reliably reproduce these errors in dev last week. I'll see if I can get him on this.

[dantswain]

This seems like it could be related: http://grokbase.com/t/kafka/users/163e1ez4q9/unknown-member-id-assigned-to-consumer-group

This error indicates that a rebalance completed successfully before the
consumer could rejoin. Basically it works like this:

1. Consumer 1 joins the group and is assigned member id A
2. Consumer 1's session timeout expires before successfully heartbeating.
3. The group is rebalanced without consumer 1.
4. Consumer 1 eventually sends its heartbeat using the old id A.
5. The coordinator rejects the heartbeat with UNKNOWN_MEMBER_ID.

[dantswain]

@CJPoll If your coworker can get this to reproduce that would be super helpful. I have been tweaking my "examples" repo and trying to change the state of brokers using docker - dantswain/kafka_ex_examples#1

[CJPoll]

So we turned off the app workers (10 BEAM VMs), and our number of connections to a single broker went from 76,000 connections to 370 (and dropping). When do TCP connections to Kafka get created, and which processes manage those?

[dantswain]

Each partition creates a connection to each broker. This is known to be wildly inefficient but generally hasn't been a problem. It's definitely in the "would like to clean that up" category, but hasn't been urgent.

[CJPoll]

I'd venture a guess that process crashes aren't closing TCP connections. I'll dig through the code to see if I can see a way that would happen.

[dantswain]

That's possible. We could be orphaning connections. The network connection code is relatively well segregated in KafkakEx. I'm not sure if that's the culprit here, though.