KAFKA-12332: Error partitions from topics with invalid IDs in LISR requests

[jolshan]

Changes how invalid IDs are handled in LeaderAndIsr requests. The ID check now occurs before leader epoch. If the ID exists and is invalid, the partition is ignored and a new INCONSISTENT_TOPIC_ID error is returned in the response.

This error should be rare, but if it occurs, it signals the need for manual intervention.

Added tests for this behavior.
I also plan to rerun the benchmark from #10071 to ensure there are no regressions with this change.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[hachikuji]

It's a minor thing, but we can avoid this nesting by restructuring the checks a little bit. For example, it would be a good idea to have a helper in Partition which encapsulates the update of the topicId state in Log. Maybe something like this:

class Partition {
  // Update topicid if necessary. 
  // Return false if the update failed because the topicId is inconsistent
  def maybeUpdateTopicId(topicId: Uuid): Boolean
}

// in ReplicaManager
if (!partition.maybeUpdateTopicId(requestTopicId)) {
  error(...)
  responseMap.put(topicPartition, Errors.UNKNOWN_TOPIC_ID)
} else if (requestLeaderEpoch > currentLeaderEpoch) {
...

[jolshan]

I was struggling with this since I didn't think it was the best solution. I'll take a look at the version you have here. Would we want to ever update the topic ID though? Maybe just a method to check if they are equal.

[jolshan]

Oh hmmm I see how this could be used on the path for actually setting it. I'll think of a good name

[jolshan]

In this case, if log is None, does it make sense to error here? If it is None, we are unable to check the topic ID

[hachikuji]

LGTM. I had one minor nitpick. Can you update the KIP for the new error code? I think it's worth sending a message to the vote thread as well to see if anyone has any concerns.

[jolshan]

@hachikuji I was wondering if I should do that. I think I've been having issues with mailer daemon trying to send to DISCUSS so I'll send to VOTE this time.

[jolshan]

Ran tests locally with newest code from trunk. Here are the tests that failed:

ConsumerBounceTest.testClose()
ConnectionQuotasTest.testListenerConnectionRateLimitWhenActualRateAboveLimit()
DynamicConnectionQuotaTest.testDynamicListenerConnectionCreationRateQuota()

These are tests I've found to be flaky running locally in the past.