KAFKA-20719: "Streams" group sticky assignor not sticky at startup

[gabriellefu]

Changes

Broker (StickyTaskAssignor)

Reported task offsets now act as a fallback previous-ownership signal.

Members reporting on-disk state for a task are registered as standby-strength candidates for it:

• appended after any real previous standbys from the target assignment
• ordered by offset sum descending (most caught-up state first)
• deduplicated by member ID
• negative sums ignored

The assignment phases are unchanged, so precedence is now:

1. current active owner
2. current standby
3. highest reported offset sum
4. least-loaded

For groups where no member reports offsets for unowned tasks, the assignment is unchanged.

Client (StreamThread)

The task-offset-sum snapshot is now populated before the consumer subscribes.

Previously it was only refreshed inside the poll loop, which races the join:

• subscribing triggers the epoch-0 heartbeat from the background thread before the first loop iteration runs
• the join heartbeat therefore carried an empty TaskOffsets list on cold start

Since offset updates do not bump the group epoch, the assignment computed after the initial rebalance delay never saw the offsets that arrived on subsequent heartbeats.

With this change, the join heartbeat reliably carries the state-directory-derived offsets, as intended by KAFKA-20116. This also benefits the warm-up refinement work, which consumes the same fields.

Testing

• Re-enables ColdStartStickinessIntegrationTest.shouldStickToLocalStateOnColdStart (added disabled in #22630)
• Both protocol variants now pass; before the fix the streams variant restored about 500 of 1000 records after a cold restart, now 0
• New unit tests in StickyTaskAssignorTest covering:
  • cold-start reclaim by member and by multi-member process
  • precedence of current active/standby owners over reported offsets, including candidate dedup
  • higher-offset-sum tie-breaking
  • negative offset sums
  • standby placement relative to the state-holding process
• Existing StickyTaskAssignorTest and StreamThreadTest suites pass unchanged