KAFKA-21214: [8/8] Add a fuzzer for the uniform2 assignor - #23724

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

…orm2 assignor

uniform2 decides how many partitions of every topic each member gets,
its quotas, in three steps, and then places the partitions. A topic
with P partitions and N subscribers has a minimum quota of P / N
partitions, and its P % N extra partitions go to as many distinct
subscribers.

This patch adds the quotas and their first two steps:
- `Quotas` holds the minimum quota and the extra partitions of every
  topic, the members getting the extra partitions, and the assignment
  sizes of the members. The steps give the extra partitions and move
  them.
- `KeepStickyExtraPartitions`: a member owning more than the minimum
  quota of a topic keeps one of its extra partitions, the members with
  the smallest assignments first when the owners outnumber the extra
  partitions.
- `AssignRemainingExtraPartitions` gives the extra partitions that
  nobody kept, one at a time, to subscribers without one of the topic,
  among the ones with the smallest assignments: the topics with the
  fewest subscribers first, and round robin within a cohort. It returns
  without allocating anything when the owners kept them all.
- `IntPair` packs two ints in a long, so that pairs sort in a `long[]`
  or a `LongHeap` without allocating.
- A step is created with what it works on and run once; the assignor
  runs the steps in order in a later patch.
- `QuotasFixture` holds the indexed group, the current assignment and
  the quotas of a group for the tests of the steps.

The balance step comes in a later patch.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

uniform2 decides how many partitions of every topic each member gets,
its quotas, in three steps: keep the sticky extra partitions, assign the
remaining ones, and balance them. The balance step moves extra
partitions from members with large assignments to members with small
ones.

This patch adds `ExtraPartitionMoves`, which the balance step uses to
choose and make these moves:
- It keeps the extra partitions of every member in two lists, those of
  topics it owns more than the minimum quota of, and the free ones.
  A transfer gives the cheapest extra partition that the receiver can
  take: one that saves a move, then one that costs none, then one that
  costs a move.
- It tells whether a receiver can take an extra partition from a giver,
  and counts the extra partitions of a giver per cohort, those of
  topics that the cohort subscribes to, to find receivers fast.
- The counts are kept for every member when they fit, and for one
  member at a time otherwise; the subscriptions are compared as bit
  sets when they fit.

The balance step comes in the next patch.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

uniform2 decides how many partitions of every topic each member gets,
its quotas, in three steps. The first two give every extra partition to
a distinct subscriber of its topic, but they can leave a member with an
assignment much larger than another subscriber of one of its topics.

This patch adds the balance step, `BalanceExtraPartitions`, a step like
the other two, which the assignor runs after them in a later patch. It
applies a single rule until it no longer applies: the member with the
largest assignment among those having an extra partition that can move
gives its cheapest one to the smallest subscriber of the topics of its
extra partitions that has no extra partition of the topic and is at
least two partitions smaller. Every move lowers the sum of the squares
of the assignment sizes, so the step ends, and it ends exactly when no
extra partition can move to a subscriber of its topic at least two
partitions smaller.

Changes:
- The givers wait in a heap, the largest first, and a giver without a
  receiver is parked until a move may give it one. A cohort sharing no
  topic with another is done at its first giver without a receiver.
- On equal assignment sizes, a giver with a free extra partition gives
  first, and a receiver owning more than the minimum quota of a topic
  with extra partitions receives first, since only it can save a move.
- When the quotas are balanced for sure, the step returns without
  building any state.
- A constructor takes the ways of counting of `ExtraPartitionMoves`, and
  a test checks that they give the same quotas on random groups.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

Once uniform2 has decided how many partitions of every topic each member
gets, its quotas, it places the partitions.

This patch adds `PartitionPlacement`, which places the partitions of a
topic on its subscribers, moving as few partitions as possible: every
owner keeps its partitions up to its quota, the lowest partition ids
first, and the other partitions go to the subscribers below their quota,
in member order. A partition thus moves only when its owner's quota is
below what it owns. The placement tells when nothing moves, so that the
topic can keep its sets of partitions.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

The uniform assignor balances the total number of partitions per member
but not the partitions of each topic: with homogeneous subscriptions, a
member can get all the partitions of a topic while the others get none
of them. The throughput is usually about the same across the partitions
of a topic but not across topics, so adding members does not spread the
load.

This patch adds uniform2, the next version of the uniform assignor, from
the steps of the previous patches. It lives next to the uniform assignor
while it is developed and is not registered as a built-in
assignor. In priority order:
- every partition of every subscribed topic goes to exactly one of its
  subscribers;
- every subscriber of a topic gets the same number of its partitions,
  the minimum quota, or one more, an extra partition;
- no extra partition could move to another subscriber of its topic whose
  assignment is at least two partitions smaller;
- partitions stay with their owners where the above allows: the
  placement moves the fewest partitions for the quotas, and an
  assignment that already has these properties is returned unchanged;
- the result only depends on the content of the input.

Changes:
- `Uniform2Assignor` and `AssignmentBuilder`, which reads the group,
  runs the `KeepStickyExtraPartitions`, `AssignRemainingExtraPartitions`
  and `BalanceExtraPartitions` steps on the quotas, places every topic
  and builds the result. Partitions of topics that a member does not
  subscribe to or which no longer exist are dropped. The class Javadoc
  of `Uniform2Assignor` describes the steps with a worked example.
- The result reuses the maps and sets of the members that do not change.
  `RangeSet` becomes public to return consecutive partitions.
  `Quotas.maxTopicsWithPartitions` bounds the size of the new maps.
- The assignor does not use racks; rack awareness comes in a later
  patch.
- `ConsumerAssignorBenchmark` gets `UNIFORM2`.
- End to end tests with hand traced expectations, written with the group
  spec and topics fixtures, and a shared checker of the properties.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

The unit tests of the uniform2 assignor check hand traced cases, but its
properties must hold for any group and any sequence of changes.

This patch adds a fuzzer checking them on random groups and random
sequences of events: members joining and leaving, one at a time or
several, topics gaining partitions, and subscriptions changing. Other
events make the input stale, as the coordinator can hand it to the
assignor: topics deleted or created again while members still hold their
partitions, and empty subscriptions.

Changes:
- `Uniform2FuzzScenario` generates the groups and the events, and hands
  every group to the assignor as a group spec built with
  `GroupSpecFixture`. `Uniform2FuzzScenarioTest` checks that a seed
  determines the scenario and that every kind of event and of stale
  input happens.
- `Uniform2AssignorFuzzTest` checks validity, spread and balance,
  determinism under shuffled inputs, the fixed point with the same map
  instances, the same map for every member whose partitions do not
  change, the movement bounds after a single leave or join, and the
  moves of tiny groups, at most one above the fewest found by a brute
  force oracle.
- It also checks that the four ways of counting of the balance step give
  the same quotas, through `AssignmentTestUtils.quotasPerWayOfCounting`.
- The default run takes seconds; environment variables scale it up, see
  the class Javadoc.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>