Hey Ryanne,
Thanks for the context, but I still don't see the situation where this
function is helpful.
Also "A's topic1 and B's a.topic1 should be the same data (minus
replication lag)." isn't true in the presence of failures/hard
restarts, compaction, and transaction markers.
Thanks,
Greg
On Wed, Jan 10, 2024 at 8:00 PM Ryanne Dolan <ryannedo...@gmail.com> wrote:
>
> > do you recall the purpose of [...] renameTopicPartition [?]
>
> A's topic1 and B's a.topic1 should be the same data (minus replication
> lag). You can't consume a record in a.topic1 that hasn't been replicated
> yet -- a remote topic by definition does not have any records that MM2
> didn't put there. So an offset for a consumer consuming from B's a.topic1
> can be translated back to an offset in A's topic1, where the data came from.
>
> Ryanne
>
> On Wed, Jan 10, 2024, 6:07 PM Greg Harris <greg.har...@aiven.io.invalid>
> wrote:
>
> > Hi Jeroen,
> >
> > I'm glad you're experimenting with MM2, and I hope we can give you
> > some more context to explain what you're seeing.
> >
> > > I wrote a small program to produce these offset syncs for the prefixed
> > > topic, and this successfully triggers the Checkpoint connector to start
> > > replicating the consumer offsets back to the primary cluster.
> >
> > This is interesting, and I wouldn't have expected it to work.
> >
> > To rewind, each flow Source->Target has a MirrorSourceConnector, an
> > Offset Syncs Topic, and a MirrorCheckpointConnector. With both
> > directions enabled, there are two separate flows each with Source,
> > Syncs topic, and Checkpoint.
> > With offset-syncs.topic.location=source, the
> > mm2-offset-syncs.b.internal on the A cluster is used for the A -> B
> > replication flow. It contains topic names from cluster A, and the
> > corresponding offsets those records were written to on the B cluster.
> > When translation is performed, the consumer groups from A are
> > replicated to the B cluster, and the replication mapping (cluster
> > prefix) is added.
> > Using your syncs topic as an example,
> > OffsetSync{topicPartition=replicate-me-0, upstreamOffset=28,
> > downstreamOffset=28} will be used to write offsets for
> > "a.replicate-me-0" for the equivalent group on the B cluster.
> >
> > When your artificial sync OffsetSync{topicPartition=a.replicate-me-0,
> > upstreamOffset=29, downstreamOffset=29} is processed, it should be
> > used to write offsets for "a.a.replicate-me-0" but it actually writes
> > offsets to "replicate-me-0" due to this function that I've never
> > encountered before: [1].
> > I think you could get those sorts of syncs into the syncs-topic if you
> > had A->B configured with offset-syncs.topic.location=source, and B->A
> > with offset-syncs-topic.location=target, and configured the topic
> > filter to do A -> B -> A round trip replication.
> >
> > This appears to work as expected if there are no failures or restarts,
> > but as soon as a record is re-delivered in either flow, I think the
> > offsets should end up constantly advancing in an infinite loop. Maybe
> > you can try that: Before starting the replication, insert a few
> > records into `a.replicate-me` to force replicate-me-0's offset n to
> > replicate to a.replicate-me-0's offset n+k.
> >
> > Ryanne, do you recall the purpose of the renameTopicPartition
> > function? To me it looks like it could only be harmful, as it renames
> > checkpoints to target topics that MirrorMaker2 isn't writing. It also
> > looks like it isn't active in a typical MM2 setup.
> >
> > Thanks!
> > Greg
> >
> > [1]:
> > https://github.com/apache/kafka/blob/13a83d58f897de2f55d8d3342ffb058b230a9183/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorCheckpointTask.java#L257-L267
> >
> > On Tue, Jan 9, 2024 at 5:54 AM Jeroen Schutrup
> > <jer...@cloudflare.com.invalid> wrote:
> > >
> > > Thank you both for your swift responses!
> > >
> > > Ryanne, the MirrorConnectorsIntegrationBaseTest only tests offset
> > > replication in cases where the producer migrated to the secondary cluster
> > > as well, starts feeding messages into the non-prefixed topic which are
> > > subsequently consumed by the consumer on the secondary cluster. After the
> > > fallback, it asserts the consumer offsets on the non-prefixed topic in
> > the
> > > secondary cluster are translated and replicated to the consumer offsets
> > of
> > > the prefixed topic in the primary cluster.
> > > In my example, the producer keeps producing in the primary cluster
> > whereas
> > > only the consumer fails over to the secondary cluster and, after some
> > time
> > > fails back to the primary cluster. This consumer will then consume
> > messages
> > > from the prefixed topic in the secondary cluster, and I'd like to have
> > > those offsets replicated back to the non-prefixed topic in the primary
> > > cluster. If you like I can provide an illustration if that helps to
> > clarify
> > > this use case.
> > >
> > > To add some context on why I'd like to have this is to retain loose
> > > coupling between producers and consumers so we're able to test failovers
> > > for individual applications without the need for all producers/consumers
> > to
> > > failover and failback at once.
> > >
> > > Digging through the Connect debug logs I found the offset-syncs of the
> > > prefixed topic not being pushed to mm2-offset-syncs.b.internal is likely
> > > the reason the checkpoint connector doesn't replicate consumer offsets:
> > > DEBUG translateDownstream(replication,a.replicate-me-0,25): Skipped
> > (offset
> > > sync not found) (org.apache.kafka.connect.mirror.OffsetSyncStore)
> > >
> > > I wrote a small program to produce these offset syncs for the prefixed
> > > topic, and this successfully triggers the Checkpoint connector to start
> > > replicating the consumer offsets back to the primary cluster.
> > > OffsetSync{topicPartition=replicate-me-0, upstreamOffset=28,
> > > downstreamOffset=28}
> > > OffsetSync{topicPartition=replicate-me-0, upstreamOffset=29,
> > > downstreamOffset=29}
> > > OffsetSync{topicPartition=a.replicate-me-0, upstreamOffset=29,
> > > downstreamOffset=29} <-- the artificially generated offset-sync
> > >
> > > At this point it goes a bit beyond my understanding of the MM2 internals
> > of
> > > whether this is a wise thing to do and if it would have any negative side
> > > effects. I'd need to spend some more time in the MM2 source, though I
> > > welcome any feedback on this hack :-)
> > >
> > > On the two complications you're mentioning Greg, the second one is
> > > something we should figure out regardless, as any given consumer group
> > may
> > > not be active on both the primary and secondary cluster as it would block
> > > MM2 from replicating its offsets from primary to the cluster-prefixed
> > topic
> > > on the secondary cluster already. On the first point, I think it would
> > be a
> > > good practice to only allow MM2 to produce to any cluster-prefixed topic
> > by
> > > using topic ACLs. In other words, the only application producing to a
> > > cluster-prefixed (or downstream) topic would be mirror maker and I think
> > > that prevents this kind of message 'drift'. In case a producer has to
> > > failover, it starts producing to the non-prefixed topic on the secondary
> > > cluster whose offsets are subject to a different Source/Checkpoint
> > > connector replication stream.
> > >
> > > On Mon, Jan 8, 2024 at 9:12 PM Ryanne Dolan <ryannedo...@gmail.com>
> > wrote:
> > >
> > > > Jeroen, MirrorClient will correctly translate offsets for both
> > failover and
> > > > failback, exactly as you describe. It's possible to automate failover
> > and
> > > > failback using that logic. The integration tests automatically fail
> > over
> > > > and fail back, for example. I've seen it done two ways: during startup
> > > > within the consumer itself, or in an external tool which writes offsets
> > > > directly. In either case MirrorClient will give you the correct
> > offsets to
> > > > resume from.
> > > >
> > > > MirrorCheckpointConnector will automatically write offsets, but only
> > under
> > > > certain conditions, to avoid accidentally overwriting offsets. I'm not
> > sure
> > > > whether you can failover and failback using just the automatic
> > behavior. My
> > > > guess is it works, but you are tripping over one of the safety checks.
> > You
> > > > might try deleting the consumer group on the source cluster prior to
> > > > failback.
> > > >
> > > > Ryanne
> > > >
> > > > On Mon, Jan 8, 2024, 9:10 AM Jeroen Schutrup
> > <jer...@cloudflare.com.invalid
> > > > >
> > > > wrote:
> > > >
> > > > > Hi all,
> > > > > I'm exploring using the MirrorSourceConnector and
> > > > MirrorCheckpointConnector
> > > > > on Kafka Connect to setup active/active replication between two Kafka
> > > > > clusters. Using the DefaultReplicationPolicy replication policy
> > class,
> > > > > messages originating from the source cluster get replicated as
> > expected
> > > > to
> > > > > the cluster-prefixed topic in the target cluster. Consumergroup
> > offsets
> > > > > from the source to target cluster are replicated likewise. However,
> > once
> > > > > the consumer group migrates from the source to the target cluster,
> > its
> > > > > offsets are not replicated from the target back to the source
> > cluster.
> > > > > For an active/active setup I'd want consumer group offsets for topic
> > > > > <source-cluster-alias>.<topic-name> in the target cluster to be
> > > > replicated
> > > > > back to <topic-name> in the source cluster. This would allow
> > consumers to
> > > > > failover & failback between clusters with minimal duplicate message
> > > > > consumption.
> > > > >
> > > > > To clarify my setup a bit; I'm running two single-broker Kafka
> > clusters
> > > > in
> > > > > Docker (cluster A & B), along with a single Connect instance on which
> > > > I've
> > > > > provisioned four source connectors:
> > > > > - A MirrorSourceConnector replicating topics from cluster A to
> > cluster B
> > > > > - A MirrorSourceConnector replicating topics from cluster B to
> > cluster A
> > > > > - A MirrorCheckpointConnector translating & replicating offsets from
> > > > > cluster A to cluster B
> > > > > - A MirrorCheckpointConnector translating & replicating offsets from
> > > > > cluster B to cluster A
> > > > >
> > > > > I'm not sure whether this is by design, or maybe I'm missing
> > something.
> > > > > I've seen a similar question posted to KAFKA-9076 [1] without a
> > > > resolution.
> > > > >
> > > > > Regards,
> > > > > Jeroen
> > > > >
> > > > > [1]
> > > > >
> > > > >
> > > >
> > https://issues.apache.org/jira/browse/KAFKA-9076?focusedCommentId=17268908&page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#comment-17268908
> > > > >
> > > >
> >
