I have a realtime table consuming messages from a 3 partition Kafka topic. Possibly due to some network issues over the weekend, all 3 consumers are repeating the same error messages about a bad offset:

2021/06/07 15:24:27.918 INFO [Fetcher] [agent_daily__2__2__20210605T0819Z] [Consumer clientId=consumer-71, groupId=] Fetch offset 22 is out of range for partition agent_daily-2, resetting offset
2021/06/07 15:24:27.919 INFO [Fetcher] [agent_daily__2__2__20210605T0819Z] [Consumer clientId=consumer-71, groupId=] Resetting offset for partition agent_daily-2 to offset 5.
2021/06/07 15:24:27.938 INFO [Fetcher] [agent_daily__1__2__20210605T0819Z] [Consumer clientId=consumer-73, groupId=] Fetch offset 20 is out of range for partition agent_daily-1, resetting offset
2021/06/07 15:24:27.939 INFO [Fetcher] [agent_daily__1__2__20210605T0819Z] [Consumer clientId=consumer-73, groupId=] Resetting offset for partition agent_daily-1 to offset 0.
2021/06/07 15:24:27.942 INFO [Fetcher] [agent_daily__0__2__20210605T0819Z] [Consumer clientId=consumer-72, groupId=] Fetch offset 24 is out of range for partition agent_daily-0, resetting offset
2021/06/07 15:24:27.943 INFO [Fetcher] [agent_daily__0__2__20210605T0819Z] [Consumer clientId=consumer-72, groupId=] Resetting offset for partition agent_daily-0 to offset 1.

2021/06/07 15:24:33.018 INFO [Fetcher] [agent_daily__2__2__20210605T0819Z] [Consumer clientId=consumer-71, groupId=] Fetch offset 22 is out of range for partition agent_daily-2, resetting offset
2021/06/07 15:24:33.018 INFO [Fetcher] [agent_daily__2__2__20210605T0819Z] [Consumer clientId=consumer-71, groupId=] Resetting offset for partition agent_daily-2 to offset 5.
2021/06/07 15:24:33.038 INFO [Fetcher] [agent_daily__1__2__20210605T0819Z] [Consumer clientId=consumer-73, groupId=] Fetch offset 20 is out of range for partition agent_daily-1, resetting offset
2021/06/07 15:24:33.039 INFO [Fetcher] [agent_daily__1__2__20210605T0819Z] [Consumer clientId=consumer-73, groupId=] Resetting offset for partition agent_daily-1 to offset 0.
2021/06/07 15:24:33.042 INFO [Fetcher] [agent_daily__0__2__20210605T0819Z] [Consumer clientId=consumer-72, groupId=] Fetch offset 24 is out of range for partition agent_daily-0, resetting offset
2021/06/07 15:24:33.043 INFO [Fetcher] [agent_daily__0__2__20210605T0819Z] [Consumer clientId=consumer-72, groupId=] Resetting offset for partition agent_daily-0 to offset 1.

The 'reset' offsets of 5, 0, and 1 are correct: I created a new 'test' table for the same topic and it used those offsets with no issue. I've tried disabling/enabling the table but it resumes those error messages. Is there some other way to reset the table consumers?

Sounds like an issue on Kafka side might have made Pinot's offset state inconsistent? https://github.com/dpkp/kafka-python/issues/972

but I can create a new table against the same topic, and not have the issue. Is there any other way to reset Pinot's offset state? Luckily this is a dev environment -- deleting and re-creating the table isn't really an option in production

If the offsets are corrupted on the kafka side, then it is a kafka side issue that needs to be prevented in production?

I can create a new table subscribed to the same topic and receive messages while the existing table still does not. If there was a Kafka issue at some point then it appears corrected

What I am saying is that Pinot stores offsets as it commits segments. If a disruptive change on kafka side makes these offsets inconsistent with what Pinot had seen earlier, then the kafka consumer inside of Pinot will run into this state.

See what the segment metadata says regarding the offset where the segment is supposed to start consuming. Pinot uses that offset. Also, have you what is your kafka retention time? Maybe increasing it a little will solve your problem

the segment metadata has the higher offsets. The consumer is trying to reset the offsets to the lower indexes, then the segment isn't using that?

So say that before the issue occured, you had offsets 1, 2, 5, 10, 20....100...500 (note they are only monotonically increasing, but don't have to be contiguous. And say pinot consumed until offset 100. Now it wants to consume from offset after 100. Howver, something disruptive happened on Kafka side, and the client is being told that there is no offset > 100.

right, but I believe Kafka knows what the new current offset should be, and the pinot consumer seems aware of it. Why wouldn't the consumer start at the new offsets: 5, 0, and 1

Fetch offset 22 is out of range for partition agent_daily-2, resetting offset
Resetting offset for partition agent_daily-2 to offset 5.
Fetch offset 20 is out of range for partition agent_daily-1, resetting offset
Resetting offset for partition agent_daily-1 to offset 0.
Fetch offset 24 is out of range for partition agent_daily-0, resetting offset
Resetting offset for partition agent_daily-0 to offset 1.

I think can only kafka know about the last consumed offset. But last consumed != last checkpointed. And will have holes between last consumed and what is checkpointed in Pinot

No, not for sure. I know the environment had network issues over the weekend so it's possible the brokers lost touch with each other and possibly their zookeeper.

I find it odd from the msg that 22 is out of range, so it is resetting to 5. Did you by any chance change the topic of consumption? Have you tried (or can you try) dropping and re-creating the table?

I made no topic changes. the consumer says it's resetting to 5, but then 30 seconds later it tries 22 again, then says resetting to 5....repeat. 5 is the correct offset and a new table with the same topic will get messages fine

@Subbu Subramaniam creating a new table works perfectly fine (so dropping and recreating should as well I think)

I have not dropped the table because I want to simulate recovering a prod environment, in which case delete the table isn't viable

Yeah, but it will pick up the offset from the checkpoint (ie previous flushed segment).

seems like more manual control is needed in this area. If a Kafka server is catastrophically lost and re-built, then that would mean any attached Pinot tables would also need rebuilt, no?

Yeah, agree it is not a great situation to be in. Which is why I want to understand what happened on the Kafka side. And if there are offset changes, is there a way for Pinot to use some info from Kafka to reliably recover. I am not sure if the 22 -> 5 offset in your example is the right mapping.

if I create a new table then it will use the reset offsets - 5 in this case

Yeah, then just delete all segments, and restart servers

that requires reload all data from the beginning though

what is reset here?

I suspect something happened with the topic such that its data was wiped. The other topics are working, so I'm not sure what happened with this particular one. In any case, if there's a catastrophic issue with the topic, I need to way to accept whatever risk and force Pinot to re-start from a particular offset. It would be on me to ensure that offset won't result in duplicate events, etc

I see.

if this was a production table with years of data and millions of events, deleting the table to start over is a huge issue

I can see that use case.

so, sort of a last resort short of deleting a table: "ignore commits, ignore offsets, etc just start consuming at this offset"

Yeah, not sure if there's a simple way already (@Subbu Subramaniam?)

There is currently no push-button way to forget kafka offsets and restart. Pinot ingestion is based on comparable offsets, with higher ones indicating newer events. All the auto-correction mechanisms also (perhaps implicitly) assume that. I am curious what the behavior of pinot is, if kafka decides to renege on the offset contract. I am wondering if any of the auto-correction mechanisms are kicking in (controller logs should tell). Restarting the controller may help. I think in your case, the controllers and servers are talking to different kafka brokers, causing inconsistency in their view. I am guessing that the auto-correction mechanism will then kick in and set the right offset in the metadata.

I opened #7039