[KIP-932]: Testing assignment routing path and timeouts

[Pratyush Ranjan (PratRanj07)]

Cover KIP-932 share-consumer assignment-routing, leader-change, and consume-flow gaps

0182-share_consumer_error_handling_mock

• test_consume_batch_retriable_error_is_flagged — a non-fatal share error surfaced via share_poll is flagged retriable (and not fatal), not just the right error code.

0183-share_consumer_leader_change_mock

• do_test_release_with_leader_change — a leader move during a RELEASE ack reroutes or surfaces NOT_LEADER_OR_FOLLOWER per partition; the RELEASE is never silently swallowed.
• do_test_rapid_leader_churn — under tight L1→L2→L3 leader churn the consumer still delivers all records, raises no fatal error, and keeps session resets bounded.

0187-share_consumer_assignment_routing

• do_test_force_precise_assignment_single_consumer — a forced manual assignment narrows a consumer to exactly the pushed partitions (not a superset/different subset).
• do_test_force_shared_partition_between_consumers — the same partition can appear in two members' assignments simultaneously (KIP-932 partition sharing).
• do_test_force_change_existing_assignment — a force-push to a different partition set drops the old and adds the new (subtract + add converge).
• do_test_force_empty_assignment — a forced empty assignment drops every partition (clear path).
• do_test_steady_state_no_spurious_serves — once steady, no new serve events fire until the assignment actually changes.
• do_test_force_reassignment_cycle_three_consumers — 3 consumers / T(6) across 4 reassignment rounds each apply the new assignment with no leakage or list growth.
• do_test_force_chaos_four_consumers — 4 consumers / T(8) across exclusive/shared/rotated rounds hold up under heavier fan-out.
• do_test_force_multi_topic_assignment_two_consumers — striped two-topic assignment resolves each topic-ID→name correctly (per-topic metadata path).
• do_test_force_same_assignment_twice_no_extra_serve — re-pushing an identical assignment triggers no extra serve (assignment-diff check).
• do_test_force_chaos_three_consumers_three_topics — 3 consumers × 3 topics across 3 reassignments (multi-topic routing under churn).
• do_test_force_chaos_four_consumers_three_topics — 4 consumers × 3 topics across 3 reassignments (heaviest membership × topic fan-out).

0188-share_consumer_consume_batch_timeout

• do_test_consume_batch_timeout_matrix — share_poll honours its timeout across the full matrix (0 / −1 / positive), returning the right counts and timing.

0189-share_consumer_decommission_mock

• do_test_decommission_while_inflight — decommissioning a broker with acks in-flight reroutes or surfaces a per-partition error (never a silent drop), and the consumer keeps working.

0191-share_consumer_consume_flow_mock

• do_test_explicit_ack_gate — in explicit mode a second share_poll before acking returns __STATE; it succeeds once acked (GAP-A).
• do_test_no_duplicate_fanout — only one fetch FANOUT is sent while a fetch is in flight (GAP-B).
• do_test_consumer_err_recovery — a CONSUMER_ERR doesn't permanently block record delivery (GAP-C).
• do_test_session_not_found_recovery — SHARE_SESSION_NOT_FOUND resets the session and records are still delivered (GAP-E).
• do_test_enqueued_flag_reset_on_reconnect — rkb_share_fetch_enqueued is reset after a broker reconnect (GAP-F).
• do_test_resubscribe_continuity — the consumer recovers and keeps delivering after unsubscribe + resubscribe (GAP-G).

[confluent-cla-assistant]

🎉 All Contributor License Agreements have been signed. Ready to merge.
_{Please push an empty commit if you would like to re-run the checks to verify CLA status for all contributors.}