More control of realtime segments - topic partitions offsets

We had some kind of issue with a Kafka topic such that the most recent offsets changed (to lower values). The realtime table is not processing any new messages to the topic now. The consumers seem to recognize what the most recent offset is, but ignore them. The following log messages repeat. If I create a new table for the same topic, it will process new messages without issue.

Short of deleting the table and re-creating, there's no way to tell Pinot to start reading from a different offset. Realtime segments seem excessively coupled to their topic partitions. A catastrophic Kafka failure resulting a a re-built Kafka server would require any attached Pinot tables to be deleted and re-built also.

I need some way to bypass the consuming segments' saved commits, offsets, etc and force it to start reading a topic partition from a specified offset. User would have to accept risks like duplicate events. Otherwise, a production issue with Kafka could require deleting Pinot tables with years of data and millions of rows.

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.

Start with Pinot realtime segments and the Kafka Fetcher behavior shown in the repeated out-of-range offset logs. Trace where consuming segments retain partition offsets and how a table currently handles offset resets. Done means users can specify a partition offset without deleting the table or its existing data, while explicitly accepting possible duplicate events.

Written by the indexing model from the issue text.