Merge pull request #849 from Aiven-Open/handle-not-leader-for-partiti…

…on-error

Retry on temporary errors from schema reader

karapace/schema_reader.py

@@ -16,8 +16,10 @@
     InvalidReplicationFactorError,
     KafkaConfigurationError,
     KafkaTimeoutError,
+    LeaderNotAvailableError,
     NoBrokersAvailable,
     NodeNotReadyError,
+    NotLeaderForPartitionError,
     TopicAlreadyExistsError,
     TopicAuthorizationFailedError,
     UnknownTopicOrPartitionError,
@@ -246,6 +248,8 @@ def _get_beginning_offset(self) -> int:
             LOG.warning("Reading begin offsets timed out.")
         except UnknownTopicOrPartitionError:
             LOG.warning("Topic does not yet exist.")
+        except (LeaderNotAvailableError, NotLeaderForPartitionError):
+            LOG.warning("Retrying to find leader for schema topic partition.")
         except Exception as e:  # pylint: disable=broad-except
             self.stats.unexpected_exception(ex=e, where="_get_beginning_offset")
             LOG.exception("Unexpected exception when reading begin offsets.")
@@ -265,6 +269,9 @@ def _is_ready(self) -> bool:
         except UnknownTopicOrPartitionError:
             LOG.warning("Topic does not yet exist.")
             return False
+        except (LeaderNotAvailableError, NotLeaderForPartitionError):
+            LOG.warning("Retrying to find leader for schema topic partition.")
+            return False
         except Exception as e:  # pylint: disable=broad-except
             self.stats.unexpected_exception(ex=e, where="_is_ready")
             LOG.exception("Unexpected exception when reading end offsets.")