KAFKA-12525: Ignoring stale status statuses when reading from Connect status topic

[vamossagar12]

During fast consecutive rebalances where a task is revoked from one worker and assigned to another one, it has been observed that there is a small time window and thus a race condition during which a RUNNING status record in the new generation is produced and is immediately followed by a delayed UNASSIGNED status record belonging to the same or a previous generation before the worker that sends this message reads the RUNNING status record that corresponds to the latest generation.
Although this doesn't inhibit the actual execution of tasks, it reports an incorrect status for those tasks(i.e UNASSIGNED). If the users have setup some kind of monitoring on tasks status then this could lead to false alarms for example.
This PR aims to solve this problem by checking if a status message is stale after reading it and updates it's status only when it is safe to. Note that it uses the same method canWriteSafely to check the staleness, so if needed, that method can be renamed (which I haven't done in this PR). Also, the original description of the ticket only talks about RUNNING/UNASSIGNED state but this PR should ideally help in filtering out all stale messages (which might still be infrequent but worth handling imo).

[vamossagar12]

@yashmayya , @gharris1727 , @C0urante Plz review whenever you get the chance. Thanks!

[yashmayya]

Thanks for the PR @vamossagar12! I have a couple of questions.

[yashmayya]

AbstractHerder::onShutdown which is responsible for updating the status to Unassigned for a task after it has been shutdown on a worker uses StatusBackingStore::putSafe. This "safe" send implementation also does the same check (whether the "newer" status update is from an older generation of tasks) on pre-existing cache entries before writing to the underlying Kafka based log. However, it looks like the cache is only updated on reading back from the log (on the KafkaBasedLog's work thread) and it isn't updated when doing the write itself. So is the cause for this bug that you're trying to fix essentially a race condition where the KafkaStatusBackingStore isn't able to read its own writes (at least immediately anyway)? If so, I wonder whether we could consider altering the design of the status backing store to update its cache on writes as well? Although I guess we could run into issues here as well since we're doing async writes so maybe it's overall better to check generations on reads through the log and ensure that the actual reflected status always corresponds to that of the latest generation. What do you think?

Edit: Never mind, I just realized that the status updates will be from different workers.

[vamossagar12]

yeah this is found when the status updates are from different workers and as you said one of them doesn't read the fresh-er status update before sending it's update. It was pretty easily reproducible by the Unit test that I had added and the status shows up.

[vamossagar12]

I did some more testing on this. One thing I would like to highlight is putSafe is also prone to letting stale updates go through from the same worker. So, the staleness checks at read time (added in this PR) can only prevent stale updates from other workers.
Note that, this changes the current behaviour when reading the write made by the put call as well i.e stale updates from other workers don't reflect anymore. This is reflective in the change I made in putConnectorStateShouldOverride.

[vamossagar12]

Thanks @yashmayya , I have responded to your comments.

[vamossagar12]

Failed tests seem unrelated.

[C0urante]

@vamossagar12 I think this approach is a bit too broad. We intentionally permit "unsafe" writes for reasons documented in the AbstractHerder and KafkaStatusBackingStore Javadocs. Specifically:

this prevents us from depending on the generation absolutely. If the group disappears and the generation is reset, then we'll overwrite the status information with the older (and larger) generation with the updated one. The danger of this approach is that slow starting tasks may cause the status to be overwritten after a rebalance has completed.

I have a few alternatives in mind for how we might address this, but haven't fully thought any of them through yet since there are several KIP PRs that need to be reviewed right now which I'm giving priority to. To give me some idea of how highly to prioritize this once those other items are taken care of, can you let me know if this issue is actively affecting you or someone you know, or if the PR is a way to get up-to-speed with Kafka Connect, or something else?

[vamossagar12]

Thanks for taking a look @C0urante . I had seen the warnings based on the Java docs and I do agree that the approach taken here is broad. IIUC, this bit

talks specifically about the route that's chosen when a connector comes up the first time. I think implicitly, it is describing the workings of the put method. Also, the pitfall of the approach
The danger of this approach is that slow starting tasks may cause the status to be overwritten after a rebalance has completed is somewhat related to this ticket.
The other bit about putSafe which talks about it being racy is also something this PR tries to address.
I do agree that the solution in this PR is a bit broader. Because the JIRA ticket talks about 2 approaches and one of them was specifically about handling just for RUNNING and UNASSIGNED state. I chose to take this route as it appeared to me that the check doesn't really harm much. I agree we still can't solve all race conditions given the kind of store we are using, but it should reduce some of the race conditions.
Regarding the priority, this is something we have noticed internally on our monitoring tools. Having said that, if you are swamped with other things, won't definitely want to burden you anymore! You can get to it whenever you have some time :)

[C0urante]

@vamossagar12 I'm a little confused by this comment:

I chose to take this route as it appeared to me that the check doesn't really harm much.

Isn't the harm here that we ignore the case brought up in the AbstractHerder Javadocs where the group can disappear and the generation is reset? With this change, we'd end up ignoring all new status updates when that happens.

I'm happy to take a look at this if the bug is coming up frequently in the wild. Will try to dive deeper next week 👍

[vamossagar12]

@C0urante , yeah I did some local testing on this. I created a connector running on a connect cluster and then brought it down and brought up a new one. And I can see that the task status are not getting updated even though the updates are being pushed to the status topic.
I guess in that case, I would keep the changes local to the issues that we have noticed and the one the JIRA mentions.

[vamossagar12]

There are some tests for which i am not sure if they are related to this change like testGetActiveTopics – org.apache.kafka.connect.integration.ConnectorTopicsIntegrationTest. Will debug them further.

[vamossagar12]

The test failures now aren't related to Connect(only one related to MM but that seems to be flaky and have notified Greg about it). @C0urante , when you have some bandwidth, would you be able to take a look at this one?

[enzo-cappa]

Wanted to to chime in an say that this issue is also affecting our deployments of Kafka Connect, in different environments. Thanks @vamossagar12 for putting this fix together. Hopefully this will be merged soon.

[vamossagar12]

@enzo-cappa , thanks.. yeah waiting for a round of review on this one.

[kkonstantine]

Thanks for the PR @vamossagar12

Even though this fix does not completely eliminate the possibility of status misrepresentation due to a race condition, the possibility of this happening is much lower now and the inconsistency should be more transient than it is today as you point out in the comments.

The test failures seem flaky and not relevant to these changes. Merging.

[nirutgupta]

@vamossagar12 @yashmayya Did we decide not to add this check for connector status? I have a customer facing an issue where the connector status transitions to UNASSIGNED after a rebalance, while the tasks continue to run fine.