1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
|
from __future__ import absolute_import
import logging
import random
import time
import kafka.common as Errors
from kafka.common import BrokerMetadata
from .future import Future
log = logging.getLogger(__name__)
class ClusterMetadata(object):
_retry_backoff_ms = 100
_metadata_max_age_ms = 300000
def __init__(self, **kwargs):
self._brokers = {}
self._partitions = {}
self._groups = {}
self._version = 0
self._last_refresh_ms = 0
self._last_successful_refresh_ms = 0
self._need_update = False
self._future = None
self._listeners = set()
for config in ('retry_backoff_ms', 'metadata_max_age_ms'):
if config in kwargs:
setattr(self, '_' + config, kwargs.pop(config))
def brokers(self):
return set(self._brokers.values())
def broker_metadata(self, broker_id):
return self._brokers.get(broker_id)
def partitions_for_topic(self, topic):
if topic not in self._partitions:
return None
return set(self._partitions[topic].keys())
def leader_for_partition(self, partition):
if partition.topic not in self._partitions:
return None
return self._partitions[partition.topic].get(partition.partition)
def coordinator_for_group(self, group):
return self._groups.get(group)
def ttl(self):
"""Milliseconds until metadata should be refreshed"""
now = time.time() * 1000
if self._need_update:
ttl = 0
else:
ttl = self._last_successful_refresh_ms + self._metadata_max_age_ms - now
retry = self._last_refresh_ms + self._retry_backoff_ms - now
return max(ttl, retry, 0)
def request_update(self):
"""
Flags metadata for update, return Future()
Actual update must be handled separately. This method will only
change the reported ttl()
"""
self._need_update = True
if not self._future or self._future.is_done:
self._future = Future()
return self._future
def topics(self):
return set(self._partitions.keys())
def failed_update(self, exception):
if self._future:
self._future.failure(exception)
self._future = None
self._last_refresh_ms = time.time() * 1000
def update_metadata(self, metadata):
# In the common case where we ask for a single topic and get back an
# error, we should fail the future
if len(metadata.topics) == 1 and metadata.topics[0][0] != 0:
error_code, topic, _ = metadata.topics[0]
error = Errors.for_code(error_code)(topic)
return self.failed_update(error)
if not metadata.brokers:
log.warning("No broker metadata found in MetadataResponse")
for node_id, host, port in metadata.brokers:
self._brokers.update({
node_id: BrokerMetadata(node_id, host, port)
})
# Drop any UnknownTopic, InvalidTopic, and TopicAuthorizationFailed
# but retain LeaderNotAvailable because it means topic is initializing
self._partitions = {}
for error_code, topic, partitions in metadata.topics:
error_type = Errors.for_code(error_code)
if error_type is Errors.NoError:
self._partitions[topic] = {}
for _, partition, leader, _, _ in partitions:
self._partitions[topic][partition] = leader
elif error_type is Errors.LeaderNotAvailableError:
log.error("Topic %s is not available during auto-create"
" initialization", topic)
elif error_type is Errors.UnknownTopicOrPartitionError:
log.error("Topic %s not found in cluster metadata", topic)
elif error_type is Errors.TopicAuthorizationFailedError:
log.error("Topic %s is not authorized for this client", topic)
elif error_type is Errors.InvalidTopicError:
log.error("'%s' is not a valid topic name", topic)
else:
log.error("Error fetching metadata for topic %s: %s",
topic, error_type)
if self._future:
self._future.success(self)
self._future = None
self._need_update = False
self._version += 1
now = time.time() * 1000
self._last_refresh_ms = now
self._last_successful_refresh_ms = now
log.debug("Updated cluster metadata version %d to %s",
self._version, self)
for listener in self._listeners:
listener(self)
def add_listener(self, listener):
"""Add a callback function to be called on each metadata update"""
self._listeners.add(listener)
def remove_listener(self, listener):
"""Remove a previously added listener callback"""
self._listeners.remove(listener)
def __str__(self):
return 'Cluster(brokers: %d, topics: %d, groups: %d)' % \
(len(self._brokers), len(self._partitions), len(self._groups))
|
