schema-reader: Shutdown service if corrupt entries in _schemas topic

Previously, when we encounter errors within the `_schemas` topic, we would continue the
message loading and skip the problematic schema.
This is not ideal as it might leave the application with corrupt schema data and the side-effects could be grave.
What we do now is to kill the service, log the errors and allow a graceful shutdown.
We will follow this work by adding metrics for such cases.

karapace/errors.py

@@ -73,3 +73,14 @@ class SubjectSoftDeletedException(Exception):
 
 class SchemaTooLargeException(Exception):
     pass
+
+
+class ShutdownException(Exception):
+    """Raised when the service has encountered an error where it should not continue and shutdown."""
+
+
+class CorruptKafkaRecordException(ShutdownException):
+    """
+    Raised when a corrupt schema is present in the `_schemas` topic. This should halt the service as
+    we will end up with a corrupt state and could lead to various runtime issues and data mismatch.
+    """

karapace/schema_reader.py

@@ -28,19 +28,20 @@
 from karapace.config import Config
 from karapace.coordinator.master_coordinator import MasterCoordinator
 from karapace.dependency import Dependency
-from karapace.errors import InvalidReferences, InvalidSchema
+from karapace.errors import CorruptKafkaRecordException, InvalidReferences, InvalidSchema, ShutdownException
 from karapace.in_memory_database import InMemoryDatabase
 from karapace.kafka.admin import KafkaAdminClient
 from karapace.kafka.common import translate_from_kafkaerror
 from karapace.kafka.consumer import KafkaConsumer
 from karapace.key_format import is_key_in_canonical_format, KeyFormatter, KeyMode
 from karapace.offset_watcher import OffsetWatcher
+from karapace.protobuf.exception import ProtobufException
 from karapace.protobuf.schema import ProtobufSchema
 from karapace.schema_models import parse_protobuf_schema_definition, SchemaType, TypedSchema, ValidatedTypedSchema
 from karapace.schema_references import LatestVersionReference, Reference, reference_from_mapping, Referents
 from karapace.statsd import StatsClient
 from karapace.typing import JsonObject, SchemaId, Subject, Version
-from karapace.utils import json_decode, JSONDecodeError
+from karapace.utils import json_decode, JSONDecodeError, shutdown
 from threading import Event, Thread
 from typing import Final, Mapping, Sequence
 
@@ -189,7 +190,7 @@ def run(self) -> None:
                 except Exception as e:  # pylint: disable=broad-except
                     LOG.exception("[Admin Client] Unexpected exception. Retrying")
                     self.stats.unexpected_exception(ex=e, where="admin_client_instantiation")
-                    self._stop_schema_reader.wait(timeout=2.0)
+                    self._stop_schema_reader.wait(timeout=KAFKA_CLIENT_CREATION_TIMEOUT_SECONDS)
 
             assert self.admin_client is not None
 
@@ -199,10 +200,11 @@ def run(self) -> None:
                     stack.enter_context(closing(self.consumer))
                 except (NodeNotReadyError, NoBrokersAvailable, AssertionError):
                     LOG.warning("[Consumer] No Brokers available yet. Retrying")
-                    self._stop_schema_reader.wait(timeout=2.0)
+                    self._stop_schema_reader.wait(timeout=KAFKA_CLIENT_CREATION_TIMEOUT_SECONDS)
                 except KafkaConfigurationError:
                     LOG.info("[Consumer] Invalid configuration. Bailing")
-                    raise
+                    self._stop_schema_reader.set()
+                    shutdown()
                 except Exception as e:  # pylint: disable=broad-except
                     LOG.exception("[Consumer] Unexpected exception. Retrying")
                     self.stats.unexpected_exception(ex=e, where="consumer_instantiation")
@@ -242,6 +244,9 @@ def run(self) -> None:
                     self.offset = self._get_beginning_offset()
                 try:
                     self.handle_messages()
+                except ShutdownException:
+                    self._stop_schema_reader.set()
+                    shutdown()
                 except Exception as e:  # pylint: disable=broad-except
                     self.stats.unexpected_exception(ex=e, where="schema_reader_loop")
                     LOG.exception("Unexpected exception in schema reader loop")
@@ -352,19 +357,17 @@ def handle_messages(self) -> None:
 
                 assert message_key is not None
                 key = json_decode(message_key)
-            except JSONDecodeError:
-                # Invalid entry shall also move the offset so Karapace makes progress towards ready state.
-                self.offset = msg.offset()
+            except JSONDecodeError as exc:
                 LOG.warning("Invalid JSON in msg.key() at offset %s", msg.offset())
-                continue
+                raise CorruptKafkaRecordException from exc
             except (GroupAuthorizationFailedError, TopicAuthorizationFailedError) as exc:
                 LOG.error(
                     "Kafka authorization error when consuming from %s: %s %s",
                     self.config["topic_name"],
                     exc,
                     msg.error(),
                 )
-                continue
+                raise ShutdownException from exc
 
             assert isinstance(key, dict)
             msg_keymode = KeyMode.CANONICAL if is_key_in_canonical_format(key) else KeyMode.DEPRECATED_KARAPACE
@@ -381,14 +384,15 @@ def handle_messages(self) -> None:
             if message_value:
                 try:
                     value = self._parse_message_value(message_value)
-                except JSONDecodeError:
-                    # Invalid entry shall also move the offset so Karapace makes progress towards ready state.
-                    self.offset = msg.offset()
+                except (JSONDecodeError, TypeError) as exc:
                     LOG.warning("Invalid JSON in msg.value() at offset %s", msg.offset())
-                    continue
+                    raise CorruptKafkaRecordException from exc
 
-            self.handle_msg(key, value)
-            self.offset = msg.offset()
+            try:
+                self.handle_msg(key, value)
+                self.offset = msg.offset()
+            except (InvalidSchema, TypeError) as exc:
+                raise CorruptKafkaRecordException from exc
 
             if msg_keymode == KeyMode.CANONICAL:
                 schema_records_processed_keymode_canonical += 1
@@ -512,9 +516,9 @@ def _handle_msg_schema(self, key: dict, value: dict | None) -> None:
 
         try:
             schema_type_parsed = SchemaType(schema_type)
-        except ValueError:
+        except ValueError as exc:
             LOG.warning("Invalid schema type: %s", schema_type)
-            return
+            raise InvalidSchema from exc
 
         # This does two jobs:
         # - Validates the schema's JSON
@@ -528,9 +532,9 @@ def _handle_msg_schema(self, key: dict, value: dict | None) -> None:
         if schema_type_parsed in [SchemaType.AVRO, SchemaType.JSONSCHEMA]:
             try:
                 schema_str = json.dumps(json.loads(schema_str), sort_keys=True)
-            except json.JSONDecodeError:
+            except json.JSONDecodeError as exc:
                 LOG.warning("Schema is not valid JSON")
-                return
+                raise InvalidSchema from exc
         elif schema_type_parsed == SchemaType.PROTOBUF:
             try:
                 if schema_references:
@@ -544,12 +548,12 @@ def _handle_msg_schema(self, key: dict, value: dict | None) -> None:
                     normalize=False,
                 )
                 schema_str = str(parsed_schema)
-            except InvalidSchema:
-                LOG.exception("Schema is not valid ProtoBuf definition")
-                return
-            except InvalidReferences:
-                LOG.exception("Invalid Protobuf references")
-                return
+            except (InvalidSchema, ProtobufException) as exc:
+                LOG.warning("Schema is not valid ProtoBuf definition")
+                raise InvalidSchema from exc
+            except InvalidReferences as exc:
+                LOG.warning("Invalid Protobuf references")
+                raise InvalidSchema from exc
 
         try:
             typed_schema = TypedSchema(
@@ -559,8 +563,8 @@ def _handle_msg_schema(self, key: dict, value: dict | None) -> None:
                 dependencies=resolved_dependencies,
                 schema=parsed_schema,
             )
-        except (InvalidSchema, JSONDecodeError):
-            return
+        except (InvalidSchema, JSONDecodeError) as exc:
+            raise InvalidSchema from exc
 
         self.database.insert_schema_version(
             subject=schema_subject,
@@ -588,13 +592,15 @@ def handle_msg(self, key: dict, value: dict | None) -> None:
                     self._handle_msg_delete_subject(key, value)
                 elif message_type == MessageType.no_operation:
                     pass
-            except (KeyError, ValueError):
+            except (KeyError, ValueError) as exc:
                 LOG.warning("The message %r-%r has been discarded because the %s is not managed", key, value, key["keytype"])
+                raise InvalidSchema("Unrecognized `keytype` within schema") from exc
 
         else:
             LOG.warning(
                 "The message %s-%s has been discarded because doesn't contain the `keytype` key in the key", key, value
             )
+            raise InvalidSchema("Message key doesn't contain the `keytype` attribute")
 
     def remove_referenced_by(
         self,

karapace/utils.py

@@ -20,6 +20,7 @@
 
 import importlib
 import logging
+import signal
 import time
 
 if importlib.util.find_spec("ujson"):
@@ -256,3 +257,11 @@ def remove_prefix(string: str, prefix: str) -> str:
         i += 1
 
     return string[i:]
+
+
+def shutdown():
+    """
+    Send a SIGTERM into the current running application process, which should initiate shutdown logic.
+    """
+    LOG.warning("=======> Sending shutdown signal `SIGTERM` to Application process <=======")
+    signal.raise_signal(signal.SIGTERM)

tests/integration/backup/test_legacy_backup.py

@@ -4,6 +4,7 @@
 Copyright (c) 2023 Aiven Ltd
 See LICENSE for details
 """
+from aiohttp.client_exceptions import ClientError
 from aiokafka.errors import InvalidTopicError
 from karapace.backup import api
 from karapace.backup.api import BackupVersion
@@ -236,9 +237,14 @@ async def test_backup_restore(
         topic_name=api.normalize_topic_name(None, config),
     )
     time.sleep(1.0)
-    res = await registry_async_client.get(f"subjects/{subject}/versions")
-    assert res.status_code == 200
-    assert res.json() == [1]
+
+    # Restoring a `v1` backup with an invalid schema stops the service as expected, but I am
+    # unsure why the logic mismatch, needs further investigation.
+    if backup_file_version == "v1":
+        with pytest.raises(ClientError):
+            await registry_async_client.get(f"subjects/{subject}/versions")
+    else:
+        await registry_async_client.get(f"subjects/{subject}/versions")
 
     _assert_canonical_key_format(
         bootstrap_servers=kafka_servers.bootstrap_servers, schemas_topic=registry_cluster.schemas_topic

tests/unit/test_schema_reader.py

@@ -10,6 +10,7 @@
 from confluent_kafka import Message
 from dataclasses import dataclass
 from karapace.config import DEFAULTS
+from karapace.errors import CorruptKafkaRecordException
 from karapace.in_memory_database import InMemoryDatabase
 from karapace.kafka.consumer import KafkaConsumer
 from karapace.key_format import KeyFormatter
@@ -199,7 +200,15 @@ def test_schema_reader_can_end_to_ready_state_if_last_message_is_invalid_in_sche
     consumer_mock = Mock(spec=KafkaConsumer)
 
     schema_str = json.dumps(
-        {"name": "init", "type": "record", "fields": [{"name": "inner", "type": ["string", "int"]}]}
+        {
+            "subject": "test",
+            "version": 1,
+            "id": 1,
+            "deleted": False,
+            "schema": json.dumps(
+                {"name": "init", "type": "record", "fields": [{"name": "inner", "type": ["string", "int"]}]}
+            ),
+        }
     ).encode()
 
     ok1_message = Mock(spec=Message)
@@ -237,16 +246,16 @@ def test_schema_reader_can_end_to_ready_state_if_last_message_is_invalid_in_sche
     schema_reader.handle_messages()
     assert schema_reader.offset == 1
     assert schema_reader.ready is False
-    schema_reader.handle_messages()
-    assert schema_reader.offset == 2
-    assert schema_reader.ready is False
-    schema_reader.handle_messages()
-    assert schema_reader.offset == 3
-    assert schema_reader.ready is False
-    schema_reader.handle_messages()  # call last time to call _is_ready()
-    assert schema_reader.offset == 3
-    assert schema_reader.ready is True
-    assert schema_reader.max_messages_to_process == MAX_MESSAGES_TO_CONSUME_AFTER_STARTUP
+
+    with pytest.raises(CorruptKafkaRecordException):
+        schema_reader.handle_messages()
+        assert schema_reader.offset == 1
+        assert schema_reader.ready is False
+
+    with pytest.raises(CorruptKafkaRecordException):
+        schema_reader.handle_messages()
+        assert schema_reader.offset == 1
+        assert schema_reader.ready is False
 
 
 def test_soft_deleted_schema_storing() -> None:

tests/unit/test_utils.py

@@ -2,7 +2,11 @@
 Copyright (c) 2024 Aiven Ltd
 See LICENSE for details
 """
-from karapace.utils import remove_prefix
+from _pytest.logging import LogCaptureFixture
+from karapace.utils import remove_prefix, shutdown
+from unittest.mock import patch
+
+import logging
 
 
 def test_remove_prefix_basic() -> None:
@@ -28,3 +32,16 @@ def test_remove_prefix_multiple_occurrences_of_prefix() -> None:
 def test_remove_prefix_empty_string() -> None:
     result = remove_prefix("", "hello ")
     assert result == ""
+
+
+def test_shutdown(caplog: LogCaptureFixture) -> None:
+    with caplog.at_level(logging.WARNING, logger="karapace.utils"):
+        with patch("karapace.utils.signal") as mock_signal:
+            mock_signal.SIGTERM = 15
+
+            shutdown()
+            mock_signal.raise_signal.assert_called_once_with(15)
+            for log in caplog.records:
+                assert log.name == "karapace.utils"
+                assert log.levelname == "WARNING"
+                assert log.message == "=======> Sending shutdown signal `SIGTERM` to Application process <======="