diff options
Diffstat (limited to 'kafka/coordinator')
| -rw-r--r-- | kafka/coordinator/base.py | 5 | ||||
| -rw-r--r-- | kafka/coordinator/consumer.py | 10 |
2 files changed, 7 insertions, 8 deletions
diff --git a/kafka/coordinator/base.py b/kafka/coordinator/base.py index c57d45a..d6ea6c0 100644 --- a/kafka/coordinator/base.py +++ b/kafka/coordinator/base.py @@ -55,9 +55,10 @@ class BaseCoordinator(object): 'heartbeat_interval_ms': 3000, 'retry_backoff_ms': 100, 'api_version': (0, 9), + 'metric_group_prefix': '', } - def __init__(self, client, metrics, metric_group_prefix, **configs): + def __init__(self, client, metrics, **configs): """ Keyword Arguments: group_id (str): name of the consumer group to join for dynamic @@ -92,7 +93,7 @@ class BaseCoordinator(object): self.heartbeat = Heartbeat(**self.config) self.heartbeat_task = HeartbeatTask(weakref.proxy(self)) self.sensors = GroupCoordinatorMetrics(self.heartbeat, metrics, - metric_group_prefix) + self.config['metric_group_prefix']) def __del__(self): if hasattr(self, 'heartbeat_task') and self.heartbeat_task: diff --git a/kafka/coordinator/consumer.py b/kafka/coordinator/consumer.py index 0429e09..a600cb4 100644 --- a/kafka/coordinator/consumer.py +++ b/kafka/coordinator/consumer.py @@ -37,10 +37,10 @@ class ConsumerCoordinator(BaseCoordinator): 'retry_backoff_ms': 100, 'api_version': (0, 9), 'exclude_internal_topics': True, + 'metric_group_prefix': 'consumer' } - def __init__(self, client, subscription, metrics, metric_group_prefix, - **configs): + def __init__(self, client, subscription, metrics, **configs): """Initialize the coordination manager. Keyword Arguments: @@ -76,9 +76,7 @@ class ConsumerCoordinator(BaseCoordinator): True the only way to receive records from an internal topic is subscribing to it. Requires 0.10+. Default: True """ - super(ConsumerCoordinator, self).__init__(client, - metrics, metric_group_prefix, - **configs) + super(ConsumerCoordinator, self).__init__(client, metrics, **configs) self.config = copy.copy(self.DEFAULT_CONFIG) for key in self.config: @@ -111,7 +109,7 @@ class ConsumerCoordinator(BaseCoordinator): self._auto_commit_task.reschedule() self.consumer_sensors = ConsumerCoordinatorMetrics( - metrics, metric_group_prefix, self._subscription) + metrics, self.config['metric_group_prefix'], self._subscription) def __del__(self): if hasattr(self, '_cluster') and self._cluster: |