fix: unknown Kafka errors with _RESOLVE error code

Also don't log KafkaUnavailableError as unexpected error

src/karapace/kafka/common.py

@@ -6,7 +6,14 @@
 from __future__ import annotations
 
 from aiokafka.client import UnknownTopicOrPartitionError
-from aiokafka.errors import AuthenticationFailedError, for_code, IllegalStateError, KafkaTimeoutError, NoBrokersAvailable
+from aiokafka.errors import (
+    AuthenticationFailedError,
+    for_code,
+    IllegalStateError,
+    KafkaTimeoutError,
+    KafkaUnavailableError,
+    NoBrokersAvailable,
+)
 from collections.abc import Iterable
 from concurrent.futures import Future
 from confluent_kafka.error import KafkaError, KafkaException
@@ -52,6 +59,8 @@ def translate_from_kafkaerror(error: KafkaError) -> Exception:
         return KafkaTimeoutError()
     if code == KafkaError._STATE:  # pylint: disable=protected-access
         return IllegalStateError()
+    if code == KafkaError._RESOLVE:  # pylint: disable=protected-access
+        return KafkaUnavailableError()
 
     return for_code(code)
 

src/karapace/schema_reader.py

@@ -11,6 +11,7 @@
     InvalidReplicationFactorError,
     KafkaConfigurationError,
     KafkaTimeoutError,
+    KafkaUnavailableError,
     LeaderNotAvailableError,
     NoBrokersAvailable,
     NodeNotReadyError,
@@ -250,6 +251,8 @@ def run(self) -> None:
                 except ShutdownException:
                     self._stop_schema_reader.set()
                     shutdown()
+                except KafkaUnavailableError:
+                    LOG.warning("Kafka cluster is unavailable or broker can't be resolved.")
                 except Exception as e:  # pylint: disable=broad-except
                     self.stats.unexpected_exception(ex=e, where="schema_reader_loop")
                     LOG.warning("Unexpected exception in schema reader loop - %s", e)