Implement KIP-345 in aiokafka (rebase of #682) #827
Conversation
suppressing lgtm warning
|
@ods Do you have suggestions on the types of tests you'd like to see here. I haven't played with this yet, just rebased it. But I'll need this functionality in the coming months so at some point I'll be going through it more thoroughly anyway. |
|
This pull request introduces 1 alert when merging 83e882a into 2c54e10 - view on LGTM.com new alerts:
|
| @@ -587,6 +588,7 @@ def _check_api_version_response(self, response): | |||
| # ((2, 6, 0), DescribeClientQuotasRequest[0]), | |||
| ((2, 5, 0), DescribeAclsRequest_v2), | |||
| ((2, 4, 0), ProduceRequest[8]), | |||
| ((2, 3, 0), SyncGroupRequest[3]), | |||
There was a problem hiding this comment.
This list is used to determine broker version. One feature is enough for each version, so this line is excessive.
Codecov Report
@@ Coverage Diff @@
## master #827 +/- ##
===========================================
- Coverage 96.96% 49.60% -47.36%
===========================================
Files 30 32 +2
Lines 5434 5497 +63
===========================================
- Hits 5269 2727 -2542
- Misses 165 2770 +2605
Flags with carried forward coverage won't be shown. Click here to find out more.
Continue to review full report at Codecov.
|
We need to proof, that this new functionality works. There must be some code with static partition assignor. |
|
I played around with testing this feature last year. I didn't make it very far, but I thought an important part would be the querying of consumers' group instance IDs – and for that, a v4 DescribeGroups req/res is necessary. kafka-python still does not have this record version. I started building it out in my branch: # aiokafka/protocol/admin.py
from kafka.protocol.admin import DescribeGroupsRequest_v3, DescribeGroupsResponse_v4
from kafka.protocol.api import Request, Response
from kafka.protocol.types import Array, Bytes, Int16, Int32, Schema, String
class DescribeGroupsResponse_v4(Response):
API_KEY = 15
API_VERSION = 4
SCHEMA = Schema(
('throttle_time_ms', Int32),
('groups', Array(
('error_code', Int16),
('group', String('utf-8')),
('state', String('utf-8')),
('protocol_type', String('utf-8')),
('protocol', String('utf-8')),
('members', Array(
('member_id', String('utf-8')),
('group_instance_id', String('utf-8')),
('client_id', String('utf-8')),
('client_host', String('utf-8')),
('member_metadata', Bytes),
('member_assignment', Bytes)))),
('authorized_operations', Int32))
)
class DescribeGroupsRequest_v4(Request):
API_KEY = 15
API_VERSION = 4
RESPONSE_TYPE = DescribeGroupsResponse_v4
SCHEMA = DescribeGroupsRequest_v3.SCHEMA |
|
We may consider removing kafka-python from dependencies, as even issues with PRs are not merged into master. |
|
Hello @rujutashinde, thank you for reminding. Unfortunately I don't have enough expertise to make this PR production ready. Would you like to participate? |
I see. Unfortunately i dont have the expertise either, i am more a user of the project. But i will give it a go when i have some time cycles. Please do update the ticket if there are any new developments to this . appreciate the quick response! |
|
Hi, |
|
Hi @tartieret, |
|
Hello! I'd love to help get this change merged into the project, my company would benefit greatly from KIP-345 mode. I'll work on the requested tests. Are tests all that is preventing this from getting merged? |
Yes, tests for it is the most important part. There is also some work to be done to resolve conflicts with merge of |
* Implement KIP-345 in aiokafka * fixing linting errors * fixing linting errors * fixing linting errors * Update tests.yml * Fix linting errors * Linting fixed, tests still failing * fixed tests. * Undoing a lot of linting * last few lints * Update assignors.py suppressing lgtm warning * fix linting * fix lgtm exception * fix trailing space * add KIP-345 tests, remove broker version check * use aiokafka AbstractPartitionAssignor * only test KIP-345 mode with valid Kafka versions * Update aiokafka/consumer/group_coordinator.py Co-authored-by: Denis Otkidach <denis.otkidach@gmail.com> * remove AbstractStaticPartitionAssignor * refactor _perform_assignment to use a JoinGroupResponse class as it's argument * poll periodically in kip-345 tests * update tests to use async_timeout * update isinstance function, use JoinGroupResponse_v5 for check --------- Co-authored-by: Vikram Patki 24489 <vpatki@wayfair.com> Co-authored-by: Vikram Patki <54442035+patkivikram@users.noreply.github.com> Co-authored-by: g-clef <spamblock@g-clef.net> Co-authored-by: Denis Otkidach <denis.otkidach@gmail.com>
|
Replaced with #941 |
Changes
Fixes #680
This is a rebase of #682 -- see #682 (comment)
Checklist
CHANGESfolder<issue_id>.<type>(e.g.588.bugfix)issue_idchange it to the pr id after creating the PR.feature: Signifying a new feature..bugfix: Signifying a bug fix..doc: Signifying a documentation improvement..removal: Signifying a deprecation or removal of public API..misc: A ticket has been closed, but it is not of interest to users.Fix issue with non-ascii contents in doctest text files.