...

• reverse.checkpointing.enabled (false) - Enables the new reverse checkpointing feature. Reverse checkpointing requires that there is an opposite, active flow, and the ReplicationPolicy implementation can tell what is the source cluster of a replica topic.

Proposed Changes

Currently, MirrorCheckpointConnector uses the offset-syncs of the same flow to generate checkpoints for downstream topics. This mechanism only allows translating offsets from upstream to downstream, and only applies to topics being replicated in the flow.

...

1. Update OffsetSyncStore to be able to track reverse offsets. (subclass/flag)
2. Update MirrorCheckpointTask to instantiate an OffsetSyncStore for the reverse checkpointing, consuming from the offset-syncs topic of the opposite flow. (i.e. if the offset syncs of the current flow are located in the source, the offset sync of the opposite flow will be in the target)
3. Update MirrorCheckpointTask to generate reverse checkpoints - i.e. in the B -> A flow, identify for all topics in B which were replicated from A (with DefaultReplicationPolicy, all topics starting with "A."), use present in the reverse offset syncs data, and transform the topic name back to the upstream name (with DefaultReplicationPolicy, strip the starting "A."), and generate a reverse checkpoint.

With the above changes, all current features and guarantees of MirrorCheckpointConnector become available for reverse checkpoints: sync groups offsets, monotonic checkpoints, offset translation based on historical offset syncs.

Based on the previous points, there are 2 hard requirements is a single hard requirement for this feature to work:

1. An opposite, active flow exists - i.e. this feature only works in a bi-directional replication.The ReplicationPolicy implementation can correctly report the source cluster of a topic (i.e. the "topicSource" method is correctly implemented).

Compatibility, Deprecation, and Migration Plan

...