Feature request or enhancement

For feature requests or enhancements, provide this information, please:

Which use case/requirement will be addressed by the proposed feature?

Epoch should be monotonically increasing. We observe cases where they are reset to zero.

Implementation ideas (optional)

Right now it can be reset to zero when heartbeats or enabled or simply when we get unlucky and there is no OFFSET_TRANSACTION_ID in the offsets. See discussion in Zulip on this. We should instead use the VGTID_KEY property which is always present in the offsets.

Will put out a PR for the fix.