node kafka consumer group not receiving message when broker down

Question

I have a topic in kafka as internal. I have created the topic using the below command

/opt/kafka/bin/kafka-topics.sh 
     --create --zookeeper zookeeper:2181 
     --replication-factor 3 -partitions 6 
     --topic internal

I need to consume all the message in three different node servers. So I am using kafka-node module as a consumer group with different consumer name. I have created a consumer group name called group1, group2, group3.

Everything working fine, I can consume all the messages in all consumers.

But when any broker is down, the consumer is not getting any message. When I list all the consumer groups, it does not showing the specific group ID.

(e.g) If nodeserver 1 is down there is no group available in broker called group1

Even if I restart the node server, it does not create any group in broker and not consuming any messages in the respective node server. But when the broker is up, and the node server is restarted, it is creating a group in broker and the node server can receive message.

consumer.js

const options = {
  kafkaHost: process.env.KAFKA_HOST, 
  groupId: group_id, //group1 (or) group2 (or) group3
  autoCommit: true,
  sessionTimeout: 15000,
  protocol: ['roundrobin'],
  fromOffset: 'latest',
  outOfRangeOffset: 'earliest',
  migrateHLC: false,
  migrateRolling: true,
  fetchMaxBytes: 1024 * 1024 * 10,
  commitOffsetsOnFirstJoin: true,
  onRebalance: (isAlreadyMember, callback) => {
      log.info({"ALREADY_MEMBER_isAlreadyMember": isAlreadyMember});
      callback();
  }
};

const consumerGroup = new ConsumerGroup(options, process.env.KAFKA_TOPIC);

// On receiving message
consumerGroup.on("message", handMessage); //handMessage is where the message has been handled

// On error receiving message
consumerGroup.on('error', function(err) {
    log.debug({"type": "KAFKA_CONSUMER_ERROR", "msg": err});
});
// On error receiving message
consumerGroup.on('offsetOutOfRange', function(err) {
    log.debug({"type": "KAFKA_CONSUMER_RANGE_ERROR", "msg": err});
});

UPDATE - 1

Even if I updated offsets.topic.replication.factor as 2 or 3, I am having the same issue. When I any broker is down, the respective node server is not consuming the message. And also when I show list of groups in broker, it shows only group2 and group3. But the group1 is not there when the broker1 is down. Even if I restart the node consumer, the group1 is not getting created.

server.properties

broker.id=1
listeners=INSIDE://:9092,OUTSIDE://:9094
advertised.listeners=INSIDE://:9092,OUTSIDE://:9094
listener.security.protocol.map=INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT
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=/kafka/kafka-logs-d3f14c9ddf0a
num.partitions=1
num.recovery.threads.per.data.dir=1
offsets.topic.replication.factor=3
transaction.state.log.replication.factor=3
transaction.state.log.min.isr=2
log.retention.hours=168
log.segment.bytes=1073741824
log.retention.check.interval.ms=300000
zookeeper.connect=zookeeper1:2181,zookeeper2:2181,zookeeper3:2181
zookeeper.connection.timeout.ms=16000
group.initial.rebalance.delay.ms=0
inter.broker.listener.name=INSIDE
advertised.port=9094
port=9092
auto.create.topics.enable=false

UPDATE - 2

When the broker is down, the group coordinator is getting deleted and it is not automatically re-electing.

Can you guys tell me what I did wrong? Or is there anything else I need to update?

Please show your Kafka server properties file... What is the replication factor of the consumer offsets topic? — OneCricketeer, Dec 24 '18 at 15:47
The `__consumer_offsets` topic is having replication factor of 1 only — Praveen Srinivasan, Jan 02 '19 at 04:59
Exactly - As I mentioned in my answer, you need to manually increase it for your cluster to be highly available — OneCricketeer, Jan 02 '19 at 16:45

Bitswazsky · Answer 1 · 2018-12-26T10:15:35.937

1

Assuming this is at-least Kafka 1.x, some changes are required regarding the HA for internal Kafka topics. Consider the following snippet from server.properties. The default values for replication are set to 1. In your case, for 3 brokers, setting these to 2 might be a good place to start.

# 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=1
transaction.state.log.replication.factor=1

Addition

As per my understanding, each consumer-group has its group coordinator. So, if there are multiple groups consuming from a topic, there can be multiple coordinators (different brokers) for that topic. A broker can act as a group coordinator for multiple consumer groups. But for a consumer-group there's only one broker which acts as coordinator. For a particular consumer group we can check which broker is the coordinator using this command:

./kafka-consumer-groups.sh --bootstrap-server <broker-host>:9092 --describe --group <consumer-group> --state

If a coordinator fails, some other broker is chosen as the coordinator. The failover strategy is explained in detail here in section 10.

edited Dec 26 '18 at 10:15

answered Dec 26 '18 at 06:03

Bitswazsky

4,242
3
29
58

Even if I updated offsets.topic.replication.factor as 2 or 3, I am having the same issue. When any broker is down, the respective node server is not consuming the message – Praveen Srinivasan Dec 26 '18 at 08:52
Can you please create a new topic with the same parameters, and try it out on that one? – Bitswazsky Dec 26 '18 at 09:40
Yes I did, but no luck – Praveen Srinivasan Dec 26 '18 at 09:53
does a broker will have only one consumer group per topic? or it can have multiple consumer group per topic? – Praveen Srinivasan Dec 26 '18 at 09:55
I got your point. But when one broker is deleted, the respective consumer group is not listed in available brokers. The group coordinator is not getting elected as other two brokers are still available. – Praveen Srinivasan Dec 26 '18 at 10:41
Even if I restart my node kafka consumer with same group name, the group name is not getting created. – Praveen Srinivasan Dec 26 '18 at 13:15

score 1 · Accepted Answer · answered Dec 26 '18 at 18:51

1

Even if I updated offsets.topic.replication.factor as 2 or 3, I am having the same issue. When I any broker is down, the respective node server is not consuming the message

After the offsets topic is created, changing this property doesn't do anything.

If it was set to one, then you now need to manually increase it

answered Dec 26 '18 at 18:51

OneCricketeer

179,855
19
132
245

Actually I have deleted the topic and re-created it, still no luck. – Praveen Srinivasan Dec 27 '18 at 09:10
Outside of a test scenario, that isn't a good idea, but I assume you did `--replication-factor` more than 1? – OneCricketeer Dec 27 '18 at 16:10
Whenever I try to test the scenario, I'll delete the topic and re-create the topic using this command `/opt/kafka/bin/kafka-topics.sh --create --zookeeper zookeeper:2181 --replication-factor 3 -partitions 6 --topic internal` – Praveen Srinivasan Dec 28 '18 at 05:20
`internal`?? The topic needs to be `__consumer_offsets` – OneCricketeer Dec 28 '18 at 17:28

node kafka consumer group not receiving message when broker down

2 Answers2