diff options
Diffstat (limited to 'kafka')
| -rw-r--r-- | kafka/consumer/group.py | 8 | ||||
| -rw-r--r-- | kafka/consumer/subscription_state.py | 5 |
2 files changed, 13 insertions, 0 deletions
diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py index f646e05..9172040 100644 --- a/kafka/consumer/group.py +++ b/kafka/consumer/group.py @@ -528,6 +528,14 @@ class KafkaConsumer(six.Iterator): log.debug("Pausing partition %s", partition) self._subscription.pause(partition) + def paused(self): + """Get the partitions that were previously paused by a call to pause(). + + Returns: + set: {partition (TopicPartition), ...} + """ + return self._subscription.paused_partitions() + def resume(self, *partitions): """Resume fetching from the specified (paused) partitions. diff --git a/kafka/consumer/subscription_state.py b/kafka/consumer/subscription_state.py index a4043a1..3d170ae 100644 --- a/kafka/consumer/subscription_state.py +++ b/kafka/consumer/subscription_state.py @@ -265,6 +265,11 @@ class SubscriptionState(object): """Return set of TopicPartitions in current assignment.""" return set(self.assignment.keys()) + def paused_partitions(self): + """Return current set of paused TopicPartitions.""" + return set(partition for partition in self.assignment + if self.is_paused(partition)) + def fetchable_partitions(self): """Return set of TopicPartitions that should be Fetched.""" fetchable = set() |