Fix StickyPartitionAssignor infinite loop with mixed subscriptions

[mstojcevich]

Fixes #3163

Problem

With mixed topic subscriptions, StickyAssignmentExecutor._perform_reassignments can loop forever. The group leader then never completes the rebalance. This affects StickyPartitionAssignor and CooperativeStickyAssignor.

Root cause

_reassign_partition_to_consumer(partition, new_consumer) may receive a substitute from PartitionMovements.get_partition_to_be_moved(). This happens when partition already moved X -> consumer and a same-topic partition moved new_consumer -> X. X owns the substitute, not consumer. Moving only the substitute leaves consumer overloaded, so the balance loop repeats the same move forever.

Fix

When the substitute's owner differs from consumer, also move partition back to that owner. Both earlier movement records cancel, so stickiness holds and no cycle is created. The net effect is one partition moved from consumer to new_consumer, which has at least 2 fewer partitions. The sum of squared assignment sizes therefore strictly decreases, and the loop terminates.

Tests

Added test_reassignment_with_mixed_subscriptions_terminates to test/consumer/test_assignors.py. It runs the reproduction from the issue and caps _move_partition calls at 1000, so a regression fails fast instead of hanging. With the fix, the assignment needs 15 moves. Without the fix, the test fails with Too many partition movements.

With the fix:

pytest test/consumer/test_assignors.py test/consumer/test_sticky_partition_movements.py
# 187 passed
pytest test/consumer
# 534 passed
make lint
# clean

I used an AI coding assistant (Claude) to investigate this and write the fix and this description.