KAFKA-13763 (2): Refactor IncrementalCooperativeAssignor for improved unit testing

[C0urante]

Jira

Builds on the changes from #11974, which exclusively touched on the IncrementalCooperativeAssignorTest test suite.

The goals here include:

1. Create an overloaded variant of the IncrementalCooperativeAssignor::performTaskAssignment method that is more testing friendly by:
   1. Returning the pre-serialization allocation and revocation of connectors and tasks across the cluster in a newly-introduced ClusterAssignment class, which eliminates the current pattern of creating a mock IncrementalCooperativeAssignor class, spying on one of its private methods, and capturing the argument passed to that spied-upon method
   2. Accepting new parameters for the current snapshot of the config topic, the last-completed generation ID, and the current generation ID, which eliminates the need to create and manage a mocked WorkerCoordinator instance during testing
   3. Not requiring parameters for the leader, config topic offset, or protocol version as these do not affect the logic for allocating connectors and tasks across a cluster
   4. Only requires a Map<String, ConnectorsAndTasks> for the set of currently-running connectors and tasks across the cluster, instead of a Map<String, ExtendedWorkerState>, which contains unnecessary information like the leader, leader URL, protocol version, and config topic offset
2. Simplify the parameter list for the IncrementalCooperativeAssignor::handleLostAssignments method, which in turn simplifies the logic for testing this class
3. Capture repeated Java 8 streams logic in simple, reusable, easily-verifiable utility methods added to the ConnectUtils class

Committer Checklist (excluded from commit message)

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

[showuon]

@C0urante , thanks for the PR. I checked the implementation twice, and test once, left some comments.

[showuon]

@C0urante , thanks for the update. Left some comments.

[C0urante]

Thanks @showuon, pushed the requested changes 👍

[showuon]

@C0urante , thanks for the tests and update. Left some comments.

[C0urante]

Thanks @showuon, good call with the improvement to the serialization logic. Took a bit of legwork but I've pushed a change that implements that and also cleans up some testing clutter; LMKWYT.

[showuon]

Left some comments.

[showuon]

Sorry, @C0urante , I was wrong. serializeMetadata will be called in metadataRequest method. And the metadataRequest is trying to create a JoinGroupRequestProtocolCollection containing all supported protocols. So if it's sessioned, we should return a collection containing sessioned, compatible, eagar. But after our change, we will return sessioned, sessioned, eagar, which is wrong. I think we should keep sessioned argument here.

[C0urante]

Ah, good catch! I've pushed a fix but tried to keep the improvement to the serialization API; if you'd prefer to go back to the sessioned boolean LMK.

[showuon]

I think we go back to sessioned in serializeMetadata method will make metadataRequest looks cleaner. Sorry that I didn't make it clear.

[C0urante]

No worries, done! Sorry for the delay.

[showuon]

LGTM! Thanks for the improvement.

[showuon]

Failed tests are unrelated.

    Build / JDK 17 and Scala 2.13 / kafka.admin.LeaderElectionCommandTest.[1] Type=Raft, Name=testPathToJsonFile, Security=PLAINTEXT
    Build / JDK 17 and Scala 2.13 / kafka.admin.LeaderElectionCommandTest.[1] Type=Raft, Name=testPathToJsonFile, Security=PLAINTEXT
    Build / JDK 8 and Scala 2.12 / kafka.server.KRaftClusterTest.testCreateClusterAndPerformReassignment()

[C0urante]

Thanks Luke!