Support producing at a specified offset with a given partition leader epoch.

Who is this for and what problem do they have today?

If we want to accurately mirror a topic from a source cluster to a destination cluster (say, in a hypothetical mm3 scenario), being able to specify an offset to produce at solves a host of problems.

• We would not need to rewrite group commits when mirroring groups (mm2 rewrites commits)
• We can have a live migration of consumers (if there is a proxy handling the consuming)
• Direct, non-group based consumers work properly

The biggie is the last bullet. Today, if a client is directly consuming a partition in a compacted, it is impossible to know what offset to start at when a topic is mirrored to a destination cluster. If the topic is non-compacted, the user can just have a mapping of "offset + src-cluster-lso", but compacted topics are gappy and the user will now know where to consume from.

If we were able to produce at a given offset with a given partition leader epoch, then we could avoid a lot of mirroring work (rewriting offsets) and support direct consumers just re-pointing at a destination cluster.

How this could theoretically work as a producer (not within the broker) is a topic configuration option where the broker trusts the producer's FirstOffset, PartitionLeaderEpoch, LastOffsetDelta, FirstTimestamp, MaxTimestamp, and in-batch Offset fields. This would allow the producer to specify these values as they are consumed exactly from the source cluster, and then mirror them exactly to the destination cluster.

To "simplify" things, we can assume that the producer will only produce strictly increasing offsets -- there are no offset rewinds. The producer will not send offset 10 and then offset 6. Note that if an epoch increases, it is valid for an offset to rewind (i.e. the client saw data loss while consuming) -- Redpanda mostly avoids this problem so we can simplify things and not worry about rewinds for a first implementation.

Also note that the producer basically controls the partition leader epoch as well, bypassing any internal partition movements (and epoch advancements) that the broker itself has already done. The broker can reject any produce requests that rewind the epoch. Also note that the producer can produce control records (i.e. mirror transactions exactly, except any open transactions).

Note that idempotency of these produce requests must still be supported.

Produce-at-offset and only be done for topics that specifically have a configuration property. The configuration property must be set at topic creation and can only be removed after that. We can also consider requiring produce requests to topics with this configuration property to have a specific (magic) request header.

What are the success criteria?

We are able to mirror a topic exactly as it exists from a source cluster to a destination cluster.

Why is solving this problem impactful?

(see top of the message)

Additional notes

This idea should not be considered if we cannot prove exact-mirroring is possible -- i.e., if there are downsides (can't support mirroring epochs / mirroring attributes / mirroring control records), then we may need to drop this idea entirely.

This idea is less compelling if we move to a future where Redpanda itself supports linking two clusters and mirroring directly (although it's possible that some aspects added for this issue are required aspects for cluster linking).

JIRA Link: CORE-7301

No source files, tests, or entry points are identified. Start by reviewing the proposal and linked JIRA CORE-7301, then establish whether producer-controlled offsets, partition leader epochs, record attributes, control records, and idempotency can support exact topic mirroring; done means proving exact mirroring is possible.