[WIP] Move offset validation logic to consumer classes"

[KKcorps]

Depends on - #12806

[Jackie-Jiang]

There is no guarantee we can always detect gap with start offset and batch first offset, especially after we introduce the exclusive start offset concept.
Seems like we still want to report gap within the general handling flow. In that case it should be easier to add an API hasGap() into MessageBatch, which can be filled by the stream consumer

[KKcorps]

There is no guarantee we can always detect gap with start offset and batch first offset, especially after we introduce the exclusive start offset concept. Seems like we still want to report gap within the general handling flow. In that case it should be easier to add an API hasGap() into MessageBatch, which can be filled by the stream consumer

That makes sense since we already have the information in the consumer from which offset did we actually wanted to consume the data vs from which offset we actually did OR any other criteria which consumer wants to implement