From 76e7d13bdd736aa23507a336d04ec025636f9404 Mon Sep 17 00:00:00 2001 From: Dana Powers Date: Sun, 10 Jan 2016 22:48:09 -0800 Subject: Check for assignment changes before yielding new record --- kafka/consumer/fetcher.py | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py index 6446f4a..91d3711 100644 --- a/kafka/consumer/fetcher.py +++ b/kafka/consumer/fetcher.py @@ -361,6 +361,15 @@ class Fetcher(six.Iterator): log.log(0, "Returning fetched records at offset %d for assigned" " partition %s", position, tp) for msg in self._unpack_message_set(tp, messages): + + # Because we are in a generator, it is possible for + # assignment to change between yield calls + # so we need to re-check on each loop + if not self._subscriptions.is_assigned(tp): + log.debug("Not returning fetched records for partition %s" + " since it is no longer assigned", tp) + break + self._subscriptions.assignment[tp].position = msg.offset + 1 yield msg else: -- cgit v1.2.1