Status

Current state: Under Discussion

Discussion thread: here

JIRA: here

Motivation

Checkpointing is a useful feature in the context of a failover: consumers can minimize duplicated message processing due to checkpoints. This is all that can be achieved in the context of a single replication flow.

...

This KIP aims to resolve this issue in bidirectional replications by allowing the checkpointing to create "reverse" checkpoints from downstream topics to upstream topics. This will help in minimizing the level of reprocessing by providing the same fine tuning options as checkpointing has, instead of having an unbounded window of reprocessing which is defined by the lag of the consumer group compared to the replication.

Public Interfaces

New configuration for MirrorCheckpointConnector:

• 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. An opposite, active flow exists - i.e. this feature only works in a bi-directional replication.
2. 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

• Since the feature is activated with a flag, which is false by default, this change is fully backward compatible, and does not require migration.

Test Plan

Integration testing on a bidirectional replication, with a failover and failback described in the Motivation section - expectation is that reprocessing will be minimal after failback.

Rejected Alternatives

1. Instead of automatically creating reverse checkpoints on all replica topics from the target cluster, a new "reverse.checkpointing.topic.filter.class" configuration can be added, with which users can specify which topics should be reverse checkpointed. While this would allow for fine grained control over the new feature, it does not seem to be useful in many cases. Additionally, in the dedicated MM2 mode, the topics filter is the same across MirrorSourceConnector and MirrorCheckpointConnector, meaning that the original checkpointing does not allow limiting the checkpointed topics to a subset of the replicated topics - adding the new reverse topic filter would introduce asymmetry between the original and the reverse checkpointing.