feat(python): let poll_messages take a consumer

[ethanlin01x]

Which issue does this PR address?

Closes #3876

Rationale

The caller could not choose a consumer, so independent Python consumers collided on one server-side offset slot.

What changed?

poll_messages built Consumer::default() internally, which resolves to numeric id 0, so every Python process polling the same topic/partition shared one offset slot and consumer groups were unreachable.

It now takes a required Consumer (Consumer.Single(id) / Consumer.Group(id)), and partition_id becomes optional so a group poll can resolve the member's assignment. Breaking change: omitting consumer raises TypeError.

Local Execution

• Passed
• Pre-commit hooks ran

AI Usage

1. Which tools? Claude code
2. Scope of usage? Analysis and implementation
3. How did you verify the generated code works correctly? Local run
4. Can you explain every line of the code if asked? Yes.

[codecov]

Codecov Report

❌ Patch coverage is 91.66667% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 82.92%. Comparing base (3cd0860) to head (d6b7d9c).

Files with missing lines	Patch %	Lines
foreign/python/src/consumer.rs	83.33%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master    #3877      +/-   ##
============================================
- Coverage     83.76%   82.92%   -0.85%     
+ Complexity     1358     1339      -19     
============================================
  Files          1212     1218       +6     
  Lines        166189   164303    -1886     
  Branches     133663   132607    -1056     
============================================
- Hits         139210   136244    -2966     
- Misses        23342    24413    +1071     
- Partials       3637     3646       +9     

Components	Coverage Δ
Rust Core	83.42% <ø> (-1.13%)	⬇️
Java SDK	66.55% <ø> (-0.13%)	⬇️
C# SDK	76.43% <ø> (+0.35%)	⬆️
Python SDK	90.12% <91.66%> (+0.12%)	⬆️
PHP SDK	84.48% <ø> (ø)
Node SDK	95.68% <ø> (-0.17%)	⬇️
Go SDK	69.04% <ø> (+0.72%)	⬆️

Files with missing lines	Coverage Δ
foreign/python/src/client.rs	99.85% <100.00%> (+<0.01%)	⬆️
foreign/python/src/identifier.rs	77.77% <ø> (ø)
foreign/python/src/lib.rs	100.00% <100.00%> (ø)
foreign/python/src/consumer.rs	82.23% <83.33%> (+0.02%)	⬆️

... and 174 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[ethanlin01x]

/request-review @slbotbm

[ethanlin01x]

/ready

[slbotbm]

There are some more tests that I would like you to add:

• Create a topic with several partitions, put a distinguishable message in each partition, join one client to the group, and repeatedly call poll_messages() without a partition_id. The client should poll all of its assigned partitions in round-robin order (assert this behaviour).
• Call poll_messages() with Consumer.Single(...) and no partition_id after sending messages to partition 0. The call should read partition 0.
• Create a group member with a known assignment and call poll_messages() with both Consumer.Group(...) and an explicitly owned partition_id. The poll should return messages from that partition.
• Add small tests for the Python-to-Rust conversion boundary (some of these could also be folded into existing tests):
  • Omitting the required consumer argument should raise TypeError.
  • Passing a value that is not a Consumer should raise TypeError.
  • An invalid string identifier should raise ValueError.
  • A negative integer or an integer larger than u32 should be rejected.
  • Valid numeric IDs should be accepted for both Consumer.Single(...) and Consumer.Group(...).
• If a client polls with Consumer.Group(group_name) without first joining that group, it has no assigned partitions and is not allowed to consume for the group. (should error)

[ethanlin01x]

Addressed each point with its own commit.

[ethanlin01x]

One note on the invalid string identifier: Consumer just holds a PyIdentifier, and the conversion only happens inside poll_messages. So Consumer.Single("") builds fine and you don't get ValueError: Invalid identifier until the poll, which is where the test catches it.

[ethanlin01x]

/ready