Status

Current state: Under Discussion

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]

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.

But in the context of bidirectional replication, the same problem occurs on failback: any progress made in the downstream topics will be lost when the consumer switches back to the upstream cluster. Depending on the lag between the consumer and the replication before failover happened, this can be potentially a large amount of reprocessing.

For example, assuming 2 clusters (A and B), consumer group C, consuming from topic T. C is originally connected to cluster A:

1. T has 1000 messages in its single partition.
2. C manages to consume the first 500 messages from T in cluster A, commits its offsets.
3. Replication correctly replicates all messages into the A.T topic in the B cluster, checkpointing creates a checkpoint close to the 500th message for group C.
4. Now a failover happens (e.g. cluster A is under maintenance), C connects to cluster B, and starts consuming from topic A.T based on the latest checkpoint.
5. C manages to consume the remaining 500 messages from A.T, commits its offsets into B.
6. Cluster A becomes available again (e.g. maintenance is done).
7. C fails back to cluster A.
8. C starts consuming from topic T. Its committed offset is still 500, meaning that C consumes the second 500 messages again.

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.

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.

In the context of a bidirectional replication, a reverse checkpoint can be created by using the offset-syncs of the opposite flow. The upstream:downstream offset pairs of the opposite flow can be used to implement the reverse mappings.

Changes to implement:

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 topics in B which were replicated from A (with DefaultReplicationPolicy, all topics starting with "A."), use reverse offset syncs, and transform the topic name back to the upstream name (with DefaultReplicationPolicy, strip the starting "A.").

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.

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.