krishan1390 opened a new pull request, #19359: URL: https://github.com/apache/pinot/pull/19359
### Problem The Kafka clients created for realtime ingestion never receive an `auto.offset.reset` value, so they fall back to Kafka's default of `latest`. Pinot seeks to an explicit start offset before every poll, so this policy only takes effect when that offset is no longer present in the log — which is exactly the case where it matters. When retention deletes records a consuming segment has not read yet, the client silently repositions to the **end** of the log, dropping every record still retained between the expired offset and the log end. A table catching up on a backlog can lose an arbitrary amount of retained data this way, and the further behind it is, the more it loses. Setting `stream.kafka.consumer.prop.auto.offset.reset` does not prevent this. Despite its name, that key is parsed into the table's initial-offset criteria (`OffsetCriteria`) and never reaches the Kafka client; its accepted values (`smallest` / `largest`) are not valid Kafka values either. ### Fix Default the Kafka client property to `earliest`, so an expired start offset resumes from the oldest retained record and only the records the broker actually deleted are skipped. The resulting gap is still surfaced by the existing `StreamDataLoss` detection. Setting the raw `auto.offset.reset` property in `streamConfigs` continues to take precedence, and the initial-offset criteria is unaffected — it still decides only where a partition with no prior offset starts consuming. ### Testing `KafkaPartitionLevelConsumerTest#testOffsetsExpired` pinned `auto.offset.reset=earliest` directly in the stream config, which masked the bug. It now uses the configuration an operator would actually write: on master it returns 0 records where 500 are expected, and it passes with this change. It also asserts that the skipped records are reported as data loss. Applied to both `pinot-kafka-3.0` and `pinot-kafka-4.0`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
