REST proxy to return 403 instead of 500 in case of deny by ACL

With REST authorization enabled, users get http 500 when attempting
to consume topics for which they are not authorized.

karapace/kafka_rest_apis/consumer_manager.py

@@ -7,7 +7,13 @@
 from collections import defaultdict, namedtuple
 from functools import partial
 from http import HTTPStatus
-from kafka.errors import GroupAuthorizationFailedError, IllegalStateError, KafkaConfigurationError, KafkaError
+from kafka.errors import (
+    GroupAuthorizationFailedError,
+    IllegalStateError,
+    KafkaConfigurationError,
+    KafkaError,
+    TopicAuthorizationFailedError,
+)
 from kafka.structs import TopicPartition
 from karapace.config import Config, create_client_ssl_context
 from karapace.kafka_rest_apis.error_codes import RESTErrorCodes
@@ -482,7 +488,7 @@ async def fetch(self, internal_name: Tuple[str, str], content_type: str, formats
                 timeout_left = max(0, (start_time - time.monotonic()) * 1000 + timeout)
                 try:
                     data = await consumer.getmany(timeout_ms=timeout_left, max_records=1)
-                except GroupAuthorizationFailedError:
+                except (GroupAuthorizationFailedError, TopicAuthorizationFailedError):
                     KarapaceBase.r(body={"message": "Forbidden"}, content_type=content_type, status=HTTPStatus.FORBIDDEN)
                 except KafkaError as ex:
                     KarapaceBase.internal_error(