KAFKA-12558: Do not prematurely mutate partiton state and provide con…

[emilnkrastev]

This PR addresses the issue described here KAFKA-12558.

Additionally, The PR also allows to configure the max outstanding syncs in MirrorSourceTask because it is currently hardcoded.
A lot of offset syncs messages are lost during burst of messages in the source cluster or when the MirrorMaker has a lot to catch up (fist run or being inactive for a while). In such scenario it will take a while to sync the offsets in the destination cluster with partitions without regular activity even with reaching the maximum parallelism - 1 task per partition.

The PR tries to mitigate the issue by providing a way to change the maximum allowed concurrent offset syncs so that there are less offset syncs loses.

Here are my steps to reproduce the offset syncs issue because of the max outstanding syncs limited to 10:

1. Source topic with 12 partitions and 1400 messages with minimal activity. Messages are getting produced on daily basis
2. Run MirrorMaker2 process within the destination cluster network with offset syncs topic location set to target and 5 tasks
3. 372 offset syncs messages arrived in the destination cluster offset syncs topic
4. 9 out of 12 partitions are not synced correctly in the destination cluster
5. Waiting for hours and more for new messages to arrive in source Kafka cluster which will sync the correct offsets

[gharris1727]

@emilnkrastev Thanks for the fix! Are you still interested in making this change?

If so, I would suggest removing the configuration changes, as those would require a KIP. We can much more easily merge a targeted bug fix.

Also I noticed that your name and email are not set in your commit. You may want to configure these so that the change can be attributed to you.

Thanks!

[emilnkrastev]

@gharris1727 I'm still interested in the change and I will update the MR in the next couple of days.

Thanks for the reply!

[emilnkrastev]

@gharris1727 The PR is updated. Could you please take a look?

[gharris1727]

Thanks for the update @emilnkrastev.

It appears there are some CI failures that mention offsets translation that we need to look into, especially in MirrorConnectorsIntegrationSSLTest

Additionally, I'm trying to think of ways to reliably test this behavior. You've translated the existing test to work with the new update signature, but we haven't really targeted the semaphore balking behavior in a regression test. The lack of code coverage is partly why we didn't notice this before.

Here's a couple of ideas that might work, i'll let you choose whichever one you think makes more sense.

• Mock the semaphore acquire/release and unit test to simulate balking in MirrorSourceTaskTest
• Mock the producer and delay the send callback to cause the real semaphore to balk in MirrorSourceTaskTest
• Refactor the MirrorSourceTask semaphore to make it more unit-testable.
• Add a test to MirrorConnectorsIntegrationTestBase with a large contention on this semaphore and assert that the checkpoint offsets are sensible.

Thanks!

[emilnkrastev]

@gharris1727 The PR has been updated by adding additional unit test for testing partition state mutation and changing the update logic. The MirrorConnectorsIntegrationSSLTest Integration test was failing due to missing synced offsets and it was fix by always updating the previous upstream/downstream offset (the original logic is doing it).

There are still test failures but I believe there are not related to the PR changes. I can see similar test failures on the other PR's CI pipelines.

[gharris1727]

Thanks so much @emilnkrastev, this LGTM once some small nits are resolved.
The test looks great, and I'm glad that we're actually covering this code path. If all of the other tests are fine with those three instance variables being null, I'm wondering how much test coverage we're really missing right now.

[gharris1727]

Nice change!

[gharris1727]

@C0urante could you take a look at this, please?

[C0urante]

Thanks for this fix, @emilnkrastev. I want to make sure I understand the goal here before commenting on the actual code changes (although I've made sure to read through them carefully before writing this).

It looks like incorrect updates to the partitionStates field (or its contents) might result in the starvation of offset syncs for some topic partitions. The task will believe that it has performed syncs that it hasn't, and as a result, may never see that the lag between the (true) last-synced offset and the offset for the last-replicated record has exceeded the value for the user-configured offset.lag.max property.

However, it looks like KAFKA-12468 and possibly KAFKA-12558 describe a different problem: it's not just a matter of offset syncs being delayed for longer than expected or even indefinitely; instead, the offset syncs that do make it to the downstream cluster actually contain incorrect values, which in some extreme cases can result in things like negative consumer lag.

I don't see how this PR would address an issue with correctness in the actual offsets used for syncs; instead, it seems like the value here is that it will allow offset syncs to take place in cases where they should be but are currently not.

Is that a fair assessment of the goal here? If so, I'm happy to review/merge a fix with that goal, and if not, would you mind shedding some light on what the goal is and where I might be missing something?

[emilnkrastev]

@C0urante your assessment of the goal is correct - the PR changes are aiming to allow syncs to take place where they should be.

I believe there are 2 different problems - one that results in delayed offset syncs and the other that results in negative lag in the destination cluster. I'm trying to mitigate the first problem (delayed offset syncs).

[C0urante]

Thanks @emilnkrastev, good to know we're on the same page!

Thinking about this a little more, I wonder if a slightly more-involved approach is warranted. With the current proposal, it looks like we might still miss some cases that are predicated on the last-seen (not last-synced) upstream/downstream offsets.

For example, if the downstream topic is deleted and then recreated, the downstreamOffset < previousDownstreamOffset part of the condition in shouldSyncOffsets will evaluate to true, and we'll try to send an offset sync as a result. If there are too many in-flight offset syncs already and we can't acquire access to the semaphore, we'll skip that sync--but since we'll also have already updated the previousDownstreamOffset field, the next call to shouldSyncOffsets won't (necessarily) return true, even though we should still attempt a sync in that case.

For a short-term fix, I think the PartitionState class could be adapted to "remember" whether syncs are necessary, and allow external callers to clear that state whenever an offset sync has actually been performed. That way, we never run the risk of "dropping" offset syncs that were supposed to be performed but were blocked by access to the semaphore.

How does that sound?

[emilnkrastev]

@C0urante Your suggestion sounds perfect.

I have updated the PR.

[C0urante]

Thanks @emilnkrastev! The functional parts LGTM; just a few thoughts about the testing strategy.

[C0urante]

Thanks @emilnkrastev! One final round of comments and this should be good to go.

[C0urante]

LGTM, thanks @emilnkrastev!