Logo Questions Linux Laravel Mysql Ubuntu Git Menu
 

Kafka INVALID_FETCH_SESSION_EPOCH

We are using a kafka broker setup with a kafka streams application that runs using Spring cloud stream kafka. Although it seems to run fine, we do get the following error statements in our log:

2019-02-21 22:37:20,253 INFO kafka-coordinator-heartbeat-thread | anomaly-timeline org.apache.kafka.clients.FetchSessionHandler [Consumer clientId=anomaly-timeline-56dc4481-3086-4359-a8e8-d2dae12272a2-StreamThread-1-consumer, groupId=anomaly-timeline] Node 2 was unable to process the fetch request with (sessionId=1290440723, epoch=2089): INVALID_FETCH_SESSION_EPOCH. 

I searched the internet but there is not much information on this error. I guessed that it could have something to do with a difference in time settings between the broker and the consumer, but both machines have the same timeserver settings.

Any idea how this can be resolved?

like image 556
mmelsen Avatar asked Feb 22 '19 09:02

mmelsen


People also ask

How to get the expected epoch of a Kafka session?

When we look into Kafka's code, we can see an example, when this is returned: if (session.epoch != expectedEpoch) { info (s"Incremental fetch session $ {session.id} expected epoch $expectedEpoch, but " + s"got $ {session.epoch}.

What is invalid_fetch_session_epoch error?

When fetch response is processed by the heartbeat thread, polling thread may send new fetch request with the same epoch as the previous fetch request if heartbeat thread hasn't yet updated the epoch. This results in INVALID_FETCH_SESSION_EPOCH error.

What is the use of fetch session epoch?

The fetch session epoch, which is used for ordering requests in a session. The topics to fetch. The name of the topic to fetch. The partitions to fetch. The partition index. The message offset.

How does a Kafka broker fetch messages from the leader?

Kafka brokers, which are replica followers, fetch messages from the leader. In order to avoid sending full metadata each time for all partitions, only those partitions which changed are sent within the same fetch session. When we look into Kafka's code, we can see an example, when this is returned:


2 Answers

There is a concept of fetch session, introduced within KIP-227 since 1.1.0 release: https://cwiki.apache.org/confluence/display/KAFKA/KIP-227%3A+Introduce+Incremental+FetchRequests+to+Increase+Partition+Scalability

Kafka brokers, which are replica followers, fetch messages from the leader. In order to avoid sending full metadata each time for all partitions, only those partitions which changed are sent within the same fetch session.

When we look into Kafka's code, we can see an example, when this is returned:

if (session.epoch != expectedEpoch) {
        info(s"Incremental fetch session ${session.id} expected epoch $expectedEpoch, but " +
          s"got ${session.epoch}.  Possible duplicate request.")
        new FetchResponse(Errors.INVALID_FETCH_SESSION_EPOCH, new FetchSession.RESP_MAP, 0, session.id)
      } else {

src: https://github.com/axbaretto/kafka/blob/ab2212c45daa841c2f16e9b1697187eb0e3aec8c/core/src/main/scala/kafka/server/FetchSession.scala#L493

In general, if you don't have thousands of partitions and, at the same time, this doesn't happen very often, then it shouldn't worry you.

like image 66
tgrez Avatar answered Jan 02 '23 06:01

tgrez


It seems as this might be caused by Kafka-8052 issue, which was fixed for Kafka 2.3.0

like image 25
Dan M Avatar answered Jan 02 '23 04:01

Dan M