From 46ebe601ef9450e948046041401f3d41c576460b Mon Sep 17 00:00:00 2001
From: Roy Antman <roy.antman@gmail.com>
Date: Mon, 8 Apr 2019 08:24:07 +0300
Subject: [PATCH 1/2] 1780 - before review

---
 kafka/consumer/fetcher.py     | 28 +++++++++++++++++++++-------
 kafka/coordinator/consumer.py |  6 +++++-
 2 files changed, 26 insertions(+), 8 deletions(-)

diff --git a/kafka/consumer/fetcher.py b/kafka/consumer/fetcher.py
index 36e269f19..4561d21ee 100644
--- a/kafka/consumer/fetcher.py
+++ b/kafka/consumer/fetcher.py
@@ -235,14 +235,16 @@ def _reset_offset(self, partition):
         log.debug("Resetting offset for partition %s to %s offset.",
                   partition, strategy)
         offsets = self._retrieve_offsets({partition: timestamp})
-        if partition not in offsets:
-            raise NoOffsetForPartitionError(partition)
-        offset = offsets[partition][0]
 
-        # we might lose the assignment while fetching the offset,
-        # so check it is still active
-        if self._subscriptions.is_assigned(partition):
-            self._subscriptions.seek(partition, offset)
+        if partition in offsets:
+            offset = offsets[partition][0]
+
+            # we might lose the assignment while fetching the offset,
+            # so check it is still active
+            if self._subscriptions.is_assigned(partition):
+                self._subscriptions.seek(partition, offset)
+        else:
+            log.debug("Could not find offset for partition %s since it is probably deleted" % (partition,))
 
     def _retrieve_offsets(self, timestamps, timeout_ms=float("inf")):
         """Fetch offset for each partition passed in ``timestamps`` map.
@@ -267,6 +269,9 @@ def _retrieve_offsets(self, timestamps, timeout_ms=float("inf")):
         start_time = time.time()
         remaining_ms = timeout_ms
         while remaining_ms > 0:
+            if not timestamps:
+                return {}
+
             future = self._send_offset_requests(timestamps)
             self._client.poll(future=future, timeout_ms=remaining_ms)
 
@@ -283,6 +288,15 @@ def _retrieve_offsets(self, timestamps, timeout_ms=float("inf")):
             if future.exception.invalid_metadata:
                 refresh_future = self._client.cluster.request_update()
                 self._client.poll(future=refresh_future, timeout_ms=remaining_ms)
+
+                # Issue #1780
+                # Recheck partition existance after after a successful metadata refresh
+                if refresh_future.succeeded() and isinstance(future.exception, Errors.StaleMetadata):
+                    log.debug("Stale metadata was raised, and we now have an updated metadata. Rechecking partition existance")
+                    unknown_partition = future.exception.args[0]  # TopicPartition from StaleMetadata
+                    if not self._client.cluster.leader_for_partition(unknown_partition):
+                        log.debug("Removed partition %s from offsets retrieval" % (unknown_partition, ))
+                        timestamps.pop(unknown_partition)
             else:
                 time.sleep(self.config['retry_backoff_ms'] / 1000.0)
 
diff --git a/kafka/coordinator/consumer.py b/kafka/coordinator/consumer.py
index b575664b2..af87a246b 100644
--- a/kafka/coordinator/consumer.py
+++ b/kafka/coordinator/consumer.py
@@ -225,7 +225,11 @@ def _on_join_complete(self, generation, member_id, protocol,
         self._subscription.needs_fetch_committed_offsets = True
 
         # update partition assignment
-        self._subscription.assign_from_subscribed(assignment.partitions())
+        try:
+            self._subscription.assign_from_subscribed(assignment.partitions())
+        except ValueError as e:
+            log.warning("%s. Probably due to a deleted topic. Requesting Re-join" % e)
+            self.request_rejoin()
 
         # give the assignor a chance to update internal state
         # based on the received assignment

From d6059aed0256b448951a6973cce2a5302b9fc1a3 Mon Sep 17 00:00:00 2001
From: Roy Antman <roy.antman@gmail.com>
Date: Mon, 8 Apr 2019 09:23:21 +0300
Subject: [PATCH 2/2] 1780 - test__reset_offset change

---
 test/test_fetcher.py | 4 ----
 1 file changed, 4 deletions(-)

diff --git a/test/test_fetcher.py b/test/test_fetcher.py
index e37a70db5..a16077152 100644
--- a/test/test_fetcher.py
+++ b/test/test_fetcher.py
@@ -138,10 +138,6 @@ def test__reset_offset(fetcher, mocker):
     fetcher._subscriptions.need_offset_reset(tp)
     mocked = mocker.patch.object(fetcher, '_retrieve_offsets')
 
-    mocked.return_value = {}
-    with pytest.raises(NoOffsetForPartitionError):
-        fetcher._reset_offset(tp)
-
     mocked.return_value = {tp: (1001, None)}
     fetcher._reset_offset(tp)
     assert not fetcher._subscriptions.assignment[tp].awaiting_reset