Raise error during rebalance if subscribed topics are not authorized

Author: dpkp

kafka/coordinator/consumer.py

@@ -136,9 +136,8 @@ def group_protocols(self):
 
     def _handle_metadata_update(self, cluster):
         # if we encounter any unauthorized topics, raise an exception
-        # TODO
-        #if self._cluster.unauthorized_topics:
-        #    raise TopicAuthorizationError(self._cluster.unauthorized_topics)
+        if cluster.unauthorized_topics:
+            raise Errors.TopicAuthorizationFailedError(cluster.unauthorized_topics)
 
         if self._subscription.subscribed_pattern:
             topics = []