Skip to content

Commit c9596ff

Browse files
dpkpdpkp
committed
Update old MetadataResponse attrs; update Admin client
1 parent 8af7cc4 commit c9596ff

3 files changed

Lines changed: 40 additions & 40 deletions

File tree

‎kafka/admin/client.py‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -554,7 +554,7 @@ def list_topics(self):
554554
A list of topic name strings.
555555
"""
556556
metadata = self._get_cluster_metadata(topics=None)
557-
return [t['topic'] for t in metadata['topics']]
557+
return [t['name'] for t in metadata['topics']]
558558

559559
def describe_topics(self, topics=None):
560560
"""Fetch metadata for the specified topics or all topics if None.
@@ -1089,9 +1089,9 @@ def _get_leader_for_partitions(self, partitions, timeout_ms=None):
10891089
valid_partitions = set()
10901090
for topic in metadata.get("topics", ()):
10911091
for partition in topic.get("partitions", ()):
1092-
t2p = TopicPartition(topic=topic["topic"], partition=partition["partition"])
1092+
t2p = TopicPartition(topic=topic["name"], partition=partition["partition_index"])
10931093
if t2p in partitions:
1094-
leader2partitions[partition["leader"]].append(t2p)
1094+
leader2partitions[partition["leader_id"]].append(t2p)
10951095
valid_partitions.add(t2p)
10961096

10971097
if len(partitions) != len(valid_partitions):

‎kafka/protocol/metadata.py‎

Lines changed: 35 additions & 35 deletions
Original file line numberDiff line numberDiff line change
@@ -12,13 +12,13 @@ class MetadataResponse_v0(Response):
1212
('port', Int32))),
1313
('topics', Array(
1414
('error_code', Int16),
15-
('topic', String('utf-8')),
15+
('name', String('utf-8')),
1616
('partitions', Array(
1717
('error_code', Int16),
18-
('partition', Int32),
19-
('leader', Int32),
20-
('replicas', Array(Int32)),
21-
('isr', Array(Int32))))))
18+
('partition_index', Int32),
19+
('leader_id', Int32),
20+
('replica_nodes', Array(Int32)),
21+
('isr_nodes', Array(Int32))))))
2222
)
2323

2424

@@ -34,14 +34,14 @@ class MetadataResponse_v1(Response):
3434
('controller_id', Int32),
3535
('topics', Array(
3636
('error_code', Int16),
37-
('topic', String('utf-8')),
37+
('name', String('utf-8')),
3838
('is_internal', Boolean),
3939
('partitions', Array(
4040
('error_code', Int16),
41-
('partition', Int32),
42-
('leader', Int32),
43-
('replicas', Array(Int32)),
44-
('isr', Array(Int32))))))
41+
('partition_index', Int32),
42+
('leader_id', Int32),
43+
('replica_nodes', Array(Int32)),
44+
('isr_nodes', Array(Int32))))))
4545
)
4646

4747

@@ -58,14 +58,14 @@ class MetadataResponse_v2(Response):
5858
('controller_id', Int32),
5959
('topics', Array(
6060
('error_code', Int16),
61-
('topic', String('utf-8')),
61+
('name', String('utf-8')),
6262
('is_internal', Boolean),
6363
('partitions', Array(
6464
('error_code', Int16),
65-
('partition', Int32),
66-
('leader', Int32),
67-
('replicas', Array(Int32)),
68-
('isr', Array(Int32))))))
65+
('partition_index', Int32),
66+
('leader_id', Int32),
67+
('replica_nodes', Array(Int32)),
68+
('isr_nodes', Array(Int32))))))
6969
)
7070

7171

@@ -83,14 +83,14 @@ class MetadataResponse_v3(Response):
8383
('controller_id', Int32),
8484
('topics', Array(
8585
('error_code', Int16),
86-
('topic', String('utf-8')),
86+
('name', String('utf-8')),
8787
('is_internal', Boolean),
8888
('partitions', Array(
8989
('error_code', Int16),
90-
('partition', Int32),
91-
('leader', Int32),
92-
('replicas', Array(Int32)),
93-
('isr', Array(Int32))))))
90+
('partition_index', Int32),
91+
('leader_id', Int32),
92+
('replica_nodes', Array(Int32)),
93+
('isr_nodes', Array(Int32))))))
9494
)
9595

9696

@@ -114,14 +114,14 @@ class MetadataResponse_v5(Response):
114114
('controller_id', Int32),
115115
('topics', Array(
116116
('error_code', Int16),
117-
('topic', String('utf-8')),
117+
('name', String('utf-8')),
118118
('is_internal', Boolean),
119119
('partitions', Array(
120120
('error_code', Int16),
121-
('partition', Int32),
122-
('leader', Int32),
123-
('replicas', Array(Int32)),
124-
('isr', Array(Int32)),
121+
('partition_index', Int32),
122+
('leader_id', Int32),
123+
('replica_nodes', Array(Int32)),
124+
('isr_nodes', Array(Int32)),
125125
('offline_replicas', Array(Int32))))))
126126
)
127127

@@ -149,15 +149,15 @@ class MetadataResponse_v7(Response):
149149
('controller_id', Int32),
150150
('topics', Array(
151151
('error_code', Int16),
152-
('topic', String('utf-8')),
152+
('name', String('utf-8')),
153153
('is_internal', Boolean),
154154
('partitions', Array(
155155
('error_code', Int16),
156-
('partition', Int32),
157-
('leader', Int32),
156+
('partition_index', Int32),
157+
('leader_id', Int32),
158158
('leader_epoch', Int32),
159-
('replicas', Array(Int32)),
160-
('isr', Array(Int32)),
159+
('replica_nodes', Array(Int32)),
160+
('isr_nodes', Array(Int32)),
161161
('offline_replicas', Array(Int32))))))
162162
)
163163

@@ -177,15 +177,15 @@ class MetadataResponse_v8(Response):
177177
('controller_id', Int32),
178178
('topics', Array(
179179
('error_code', Int16),
180-
('topic', String('utf-8')),
180+
('name', String('utf-8')),
181181
('is_internal', Boolean),
182182
('partitions', Array(
183183
('error_code', Int16),
184-
('partition', Int32),
185-
('leader', Int32),
184+
('partition_index', Int32),
185+
('leader_id', Int32),
186186
('leader_epoch', Int32),
187-
('replicas', Array(Int32)),
188-
('isr', Array(Int32)),
187+
('replica_nodes', Array(Int32)),
188+
('isr_nodes', Array(Int32)),
189189
('offline_replicas', Array(Int32)))),
190190
('authorized_operations', BitField))),
191191
('authorized_operations', BitField)

‎test/integration/test_admin_integration.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -406,8 +406,8 @@ def test_create_delete_topics(kafka_admin_client):
406406
@pytest.mark.skipif(env_kafka_version() < (2, 2), reason="Leader Election requires broker >=2.2")
407407
def test_perform_leader_election(kafka_admin_client, topic):
408408
topic_metadata = kafka_admin_client.describe_topics([topic])[0]
409-
assert topic_metadata['topic'] == topic
410-
partitions = list(map(lambda p: p['partition'], topic_metadata['partitions']))
409+
assert topic_metadata['name'] == topic
410+
partitions = list(map(lambda p: p['partition_index'], topic_metadata['partitions']))
411411
election_type = 0 # Preferred
412412
topic_partitions = {topic: partitions}
413413
# When Leader Election is not needed (cluster is stable), error 84 is returned

0 commit comments

Comments
 (0)