Kafka Consumer

Question 1

What is a consumer?

Answer 1

A consumer is any application which consumes data from Kafka topic by subscribing to it.

Question 2

Can Kafka consumers subscribe to multiple topics?

Answer 2

Yes a single Kafka consumer can subscribe to multiple topics.

Question 3

Why do we need Consumer groups?

Answer 3

Suppose producer is producing too much data and single consumer is not able to keep up with the demand to process all data, we spin up multiple consumers and keep them in same logical group called Consumer Group. Consumer group share the same offset id. So once a record has been read by one consumer, other cosumer in the same consumer group will not get that message. This allows sharing of load.

Question 4

Do consumers in Consumer Group share offset?

Answer 4

Yes, consumers in same consumer group share offset id.

Question 5

How many consumers can consume from a partition?

Answer 5

Only one consumer can consume from one partition.

Question 6

What is the relation between number of partitions and number of consumers?

Answer 6

The number of consumers can be either equal to or less than number of partitions. So only one consumer will be updating the offset metadata at a time. One consumer may be handling either one or more than one partitions at a time. But two consumers cannot handle single partition.

Question 7

At which level is the metadata about the offset maintained? Consumer or Consumer Group level?

Answer 7

The metadata about the offset is maintained at the Consumer group level, so if same message is to be consumed by multiple consumers then they need to be part of different consumer groups.

Question 8

Which property defines the consumer group that the consumer will belong to?

Answer 8

“group.id” property defines the group that the consumer will join.

Question 9

What will happen if you only have few consumers consuming from multiple partitions?

Answer 9

There will be lag in processing of messages, as the consumers cannot keep up with the rate at which messages are being published.

Question 10

Is it possible to give partition number to consumer when starting?

Answer 10

Yes it is possible to give consumer partition number with topic.

Question 11

What will happen if you have a topic with one partition only and the consumer group has three consumers with same consumer group name?

Answer 11

In that case, only one consumer will be able to consume the data, other consumers will not get data.

Question 12

What is the best practice for number of consumers in consumer group supposing that the topic has n partitions?

Answer 12

It is a best practice to have n + 1 or n + 2 consumers as they will act as failovers when any consumer goes down.

Question 13

What is rebalancing in Kafka and why is it required?

Answer 13

Moving ownership of a parition from one consumer to another is called rebalance.
Whenever a new consumer joins the group or a consumer goes dead, new partition is added, then Kafka will try to balance the load among all the consumers. This process of redistributing the load is called rebalancing.

Question 14

When can rebalancing occur?

Answer 14

1) A new consumer is added
2) Consumer crashes or is DEAD
3) Consumer is not sending heartbeat response due to doing some heavy processing, it is deemed as logically DEAD
4) A new partition is added

Question 15

What affect does existing consumers have when rebalance occurs?

Answer 15

Consumers loose their current state, ie. which partition is currently assigned to them and such. During rebalance, no messages will be processed from the partitions that owned by the dead consumer.

Question 16

How does heartbeat work in Kafka, who monitors the heartbeat?

Answer 16

Consumers connect to group co-ordinators and send heart beat on some regular intervals. If consumer stops sending heartbeats for long enough, its session will timeout and the group co-ordinator will consider it dead and trigger a rebalance.

Question 17

What is the relation between Group Co-ordinator and topic?

Answer 17

For every topic a Group co-ordinator is assigned.

Question 18

What is consumer leader?

Answer 18

The first consumer that joins the consumer group becomes the consumer leader for that group.

Question 19

What is a JoinGroup request?

Answer 19

When a consumer wants to join a group, it sends a JoinGroup request to the group co-ordinator.

Question 20

What are the responsibilities of consumer leader?

Answer 20

1) The consumer leader receives list of all consumers in the consumer group from group co-ordinator.
2) Leader is responsible for assigning a subset of partitions to each consumer.
3) After deciding the partition assignment the leader will send all this information to group co-ordinator
4) Group co-ordinator sends this information to all the consumers.
This process is repeated every time rebalance is triggered.

Question 21

What is use of interface PartitionAssignor?

Answer 21

Consumer leader uses implementation of PartitionAssignor interface to decide which partitions should be handled by which consumer.

Question 22

Which consumer has all the information about parition assignment regarding consumer group?

Answer 22

The consumer leader is the only client process that has full list of consumers in the group and its assignments. All the other consumers in the group only have knowledge about their assignments.

Question 23

What happens when we call poll() method for the first time in consumer code?

Answer 23

First time we call poll with a new consumer, it is responsible for finding the GroupCoordinator, joining the consumer group and receiving the partition assignment.
If rebalance is triggered it will also be handled in poll method.

Question 24

Which things are handled in poll method?

Answer 24

1) Joining consumer group
2) Contacting Group coordinator
3) Getting the assigned paritions
4) Rebalancing

So it handles co-ordination, parition rebalances, data fetching.

Question 25

Which method is used to fetch data from Kafka in Consumer?

Answer 25

poll() method is used.

Question 26

What does poll() method return?

Answer 26

poll() method returns ConsumerRecords which contains multiple records. Then we can iterate through the ConsumerRecord using foreach loop as ConsumerRecords implements Iterable interface.

Question 27

What does ConsumerRecord contain?

Answer 27

It contains offset, partition, topic, key, value.

Question 28

What is the use fetch.min.bytes configuration property?

Answer 28

When you call poll() method, it specifies how many min bytes needs to be collected before the poll method returns.

Question 29

What is the use of fetch.max.wait.ms configuration property?

Answer 29

The maximum amount to time the server will block before answering the fetch request if there isn’t sufficient data to immediately satisfy the requirement given by fetch.min.bytes.

Question 30

What is max.partition.fetch.bytes configuration property?

Answer 30

It controls the maximum number of bytes the server will return per partition. The default is 1 MB.

Question 31

What is auto.offset.reset configuration property?

Answer 31

This property controls the behavior of the consumer when it starts reading a partition for whih it doesn’t have a committed offset or if the committed offset it has is invalid.
Two possible values are earliest, latest.

Question 32

What is enable.auto.commit configuration property?

Answer 32

It controls whether the consumer will commit offsets automatically. It defaults to true.

Question 33

What is session.timeout.ms configuration property?

Answer 33

The amount of time the consumer can be out of contact with brokers while still considered alive defaults to 3 seconds.

Question 34

What is max.poll.records configuration property?

Answer 34

The maximum number of records that a single call to poll() will return. This is useful to help control the amount of data your application will need to process in the polling loop.

Question 35

What is client.id configuration property?

Answer 35

This can be any string and will be used by the brokers to identify messages sent from the client. It is sued in logging and metrics and for quotas.

Question 36

How does Kafka know the state of the consumers, as to which consumer has read how many messages from which partition, basically the offset it?

Answer 36

Consumers need to update Kafka about the progress. That is know Kafka will come to know about the status. This action of updating the current position in the partition is called commit.

Question 37

How is Kafka different from JMS in regards to offset?

Answer 37

In JMS the messaging system keeps track of received messages. But in case of Kafka consumer keeps track of received messages and just updates Kafka regarding their status.

Question 38

In how many different ways can Consumer commit?

Answer 38

1) commitSync - Till the time commit is not done, your processing logic halts - Commit current offset
2) commitAsync - Commit happens in different thread
3) Automatic commit - if we enable enable.auto.commit then every 5 seconds the consumer will commit the largest offset your client received from poll()
4) Combining sync and async commits
5) Commit specified offset

Question 39

Which problems can occur if offset is not committed properly?

Answer 39

Either the consumer will lose some messages, or the consumer will reprocess some messages.

Question 40

What does auto.commit.interval.ms configuration property do?

Answer 40

It specifies the amount of time after which commit will be done. It is driven by poll method. Everytime consumer consumes using poll method, it checks to see if it is time to commit, if so it will commit the offsets it returned in the last poll.
close() methods commits offsets automatically.

Question 41

What are drawbacks of commitSync or commit current offset technique?

Answer 41

1) Application is blocked until the broker responds
2) This limits the throughput of the application.
3) Throughput can be improved by less frequent commits but it increases the chances of potential duplicates.

Question 42

How to get feedback about async commit whether it succeeded or not?

Answer 42

In consumer.commitAsync(OffsetCommitCallback), we can provide implementation of OffsetCommitCallback. It has onComplete(Map offsets, Exception ex) where you get exception non null if there was any issue.

Question 43

What solutions do we have when async offset commit fails?

Answer 43

We can either retry to commit the same offset or commit the latest offset.

Question 44

Which is the best practice to handle shutdown when using commitAsync?

Answer 44

The best practice is to combine async and sync. Before the consumer shuts down, the sync will commit the latest offset before exiting.

Question 45

Is it possible to commit specified offset?

Answer 45

Yes using commitAsync we can also provide Map into the commit method.

Question 46

What is the problem that can occur with commits when partition rebalance takes place due to a new consumer etc.

Answer 46

When partition rebalance takes place the consumers pause for a while before the partitions are distributed among them. So we need to commit offset before that. How do we know when rebalancing is about to take place. We can impement listener called ConsumerRebalanceListener.

Question 47

Which methods are there in ConsumerRebalanceListener?

Answer 47

onPartitionsRevoked() - Called when a new consumer is added and before rebalancing starts and current consumer has stopped consuming messages.

onPartitionAssigned() - Called after the partitions have been reassigned to the broker but before the consumer starts consuming messages.

Question 48

Sample code to provide rebalance listener to consumer?

Answer 48

consumer.subscribe(topicList, new RebalanceListener());

Question 49

By default poll method starts consuming messages from where start or last?

Answer 49

poll() method starts consuming messages from the last committed offset in each partitions and process all messages in sequence.

Question 50

In how many ways can we consumer records with specific offset?

Answer 50

1) Skip to beginning - It will start reading messages from the beggining of partition.
consumer. seekToBeginning(TopicPartition tp);
2) Skip to end - If we want to skip all the way to the end of the partition and consume only the new messages.
consumer. seekToEnd(TopicPartition tp);

Question 51

If we commit the offset after processing the message but still there is chance that the application will crash after the message was processed and stored in database (i.e. side effect is done) but we are not able to update Kafka. How to solve this issue?

Answer 51

1) If the problem is due to partition rebalance then we can implement ConsumerRebalanceListener and save the offset to db. and when partition is assigned back we can read the offsets from database.

Question 52

Which method is used to change consumer offset to a particular value from which it will start processing messages?

Answer 52

consumer.seek(topicPartition, offset) can be used to change the consumer offset to a particular value.

Question 53

How to safely break out of poll loop? Consumer is waiting for new messages but now you want to shutdown how to break from that safely?

Answer 53

We can do consumer.wakeup() in the ShutdownHook. ShutdownHook is run in another thread. So when we call consumer.wakeup() from shutdown hook, then we will get WakeupException from poll() method. This is signal that someone called wakeup on consumer.

Question 54

What happens if there is single partition in topic and multiple consumers try to connect to same topic and consume messages (Either as two separate processes or two different threads)

Answer 54

At a time only one consumer is able to read messages from partition. Once one consumer disconnects, other is allowed to read messages from partition. Ensuring ordered delivery of messages.