diff --git a/kafka/cluster.py b/kafka/cluster.py
index fd5abe852..9d2115859 100644
--- a/kafka/cluster.py
+++ b/kafka/cluster.py
@@ -112,6 +112,7 @@ def partitions_for_topic(self, topic):
 
         Returns:
             set: {partition (int), ...}
+            None if topic not found.
         """
         if topic not in self._partitions:
             return None
diff --git a/kafka/consumer/group.py b/kafka/consumer/group.py
index fa5b8ea8b..71b295d49 100644
--- a/kafka/consumer/group.py
+++ b/kafka/consumer/group.py
@@ -634,7 +634,7 @@ def partitions_for_topic(self, topic):
         if partitions is None:
             self._fetch_all_topic_metadata()
             partitions = cluster.partitions_for_topic(topic)
-        return partitions
+        return partitions or set()
 
     def poll(self, timeout_ms=0, max_records=None, update_offsets=True):
         """Fetch data from assigned topics / partitions.
diff --git a/kafka/coordinator/consumer.py b/kafka/coordinator/consumer.py
index 4c1b38644..3d180ca0c 100644
--- a/kafka/coordinator/consumer.py
+++ b/kafka/coordinator/consumer.py
@@ -203,8 +203,8 @@ def _auto_assign_all_partitions(self):
     def _build_metadata_snapshot(self, subscription, cluster):
         metadata_snapshot = {}
         for topic in subscription.group_subscription():
-            partitions = cluster.partitions_for_topic(topic) or []
-            metadata_snapshot[topic] = set(partitions)
+            partitions = cluster.partitions_for_topic(topic)
+            metadata_snapshot[topic] = partitions or set()
         return metadata_snapshot
 
     def _lookup_assignor(self, name):