KAFKA-13769 Fix version check in SubscriptionStoreReceiveProcessorSupplier

[Gerrrr]

This patch fixes another incorrect version check in the FK code and adds unit tests that would have caught this bug.

Committer Checklist (excluded from commit message)

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

[Gerrrr]

@vvcephei Can you please review?

[Gerrrr]

Test failure:

[2022-08-18T14:24:17.527Z] FetchRequestTestDowngrade > testTopicIdIsRemovedFromFetcherWhenControllerDowngrades() STARTED
[2022-08-18T14:24:17.527Z] integration.kafka.server.FetchRequestTestDowngrade.testTopicIdIsRemovedFromFetcherWhenControllerDowngrades() failed, log available in /home/jenkins/jenkins-agent/workspace/Kafka_kafka-pr_PR-12535/core/build/reports/testOutput/integration.kafka.server.FetchRequestTestDowngrade.testTopicIdIsRemovedFromFetcherWhenControllerDowngrades().test.stdout
[2022-08-18T14:24:17.527Z] 
[2022-08-18T14:24:17.527Z] FetchRequestTestDowngrade > testTopicIdIsRemovedFromFetcherWhenControllerDowngrades() FAILED
[2022-08-18T14:24:17.527Z]     org.apache.kafka.common.errors.TopicExistsException: Topic '__consumer_offsets' already exists.

seems unrelated.

[vvcephei]

Thanks, @Gerrrr !

Since we've just been burned by the specificity of unit testing, do you think there's value in adding an upgrade test? There are good examples of upgrade tests in the system tests, if you can't immediately see how to do it in an integration test.

We can also do that in follow-up if you don't want to block this fix, but it's starting to look like a good idea for preventing regressions in the future.

[vvcephei]

I assume that you've now searched the code base for every usage of getVersion and CURRENT_VERSION to make sure we don't have any more sneaky bugs, right?

[Gerrrr]

There is

kafka/streams/src/main/java/org/apache/kafka/streams/kstream/internals/foreignkeyjoin/SubscriptionResolverJoinProcessorSupplier.java

Lines 85 to 91 in 2ff4c0a

	public void process(final Record<K, SubscriptionResponseWrapper<VO>> record) {
	if (record.value().getVersion() != SubscriptionResponseWrapper.CURRENT_VERSION) {
	//Guard against modifications to SubscriptionResponseWrapper. Need to ensure that there is
	//compatibility with previous versions to enable rolling upgrades. Must develop a strategy for
	//upgrading from older SubscriptionWrapper versions to newer versions.
	throw new UnsupportedVersionException("SubscriptionResponseWrapper is of an incompatible version.");
	}

I did not update it because the version check there is for SubscriptionResponseWrapper and its version wasn't incremented.

[vvcephei]

Ok, let's make sure we follow-up with that fix as well.

[Gerrrr]

I added an upgrade test back in #12122. Unfortunately, it didn't catch these version check bugs. I am going to figure out why and improve the test. I'd prefer to do that in a separate patch though.

[vvcephei]

Test failure was unrelated:

[Build / JDK 11 and Scala 2.13 / integration.kafka.server.FetchRequestTestDowngrade.testTopicIdIsRemovedFromFetcherWhenControllerDowngrades()](https://ci-builds.apache.org/job/Kafka/job/kafka-pr/job/PR-12535/1/testReport/junit/integration.kafka.server/FetchRequestTestDowngrade/Build___JDK_11_and_Scala_2_13___testTopicIdIsRemovedFromFetcherWhenControllerDowngrades__/)