Merge branch 'main' into avro_references2

.github/workflows/lint.yml

@@ -23,7 +23,7 @@ jobs:
     - uses: actions/setup-python@v5
       with:
         cache: pip
-        python-version: '3.11'
+        python-version: '3.12'
     - name: Install libsnappy-dev
       run: sudo apt install libsnappy-dev
     # required for pylint
@@ -42,7 +42,7 @@ jobs:
     - uses: actions/setup-python@v5
       with:
         cache: pip
-        python-version: '3.11'
+        python-version: '3.12'
     - name: Install libsnappy-dev
       run: sudo apt install libsnappy-dev
     - run: pip install -r requirements/requirements.txt -r requirements/requirements-typing.txt

.github/workflows/schema.yml

@@ -15,7 +15,7 @@ jobs:
           fetch-depth: 0
       - uses: actions/setup-python@v5
         with:
-          python-version: "3.11"
+          python-version: "3.12"
           cache: pip
           cache-dependency-path:
             requirements.txt

.github/workflows/tests.yml

@@ -20,7 +20,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: [ '3.8', '3.9', '3.10', '3.11' ]
+        python-version: [ '3.8', '3.9', '3.10', '3.11', '3.12' ]
     env:
       PYTEST_ADDOPTS: >-
         --log-dir=/tmp/ci-logs

.pre-commit-config.yaml

@@ -21,7 +21,7 @@ repos:
     entry: ./copyright.sh
 
 - repo: https://github.com/shellcheck-py/shellcheck-py
-  rev: v0.9.0.2
+  rev: v0.10.0.1
   hooks:
   - id: shellcheck
 
@@ -56,7 +56,7 @@ repos:
   - id: black
 
 - repo: https://github.com/PyCQA/flake8
-  rev: 6.0.0
+  rev: 7.1.0
   hooks:
   - id: flake8
 
@@ -73,6 +73,6 @@ repos:
 - repo: https://github.com/PyCQA/pylint
   # Note: pre-commit autoupdate changes to an alpha version. Instead, manually find the
   # latest stable version here: https://github.com/pylint-dev/pylint/releases
-  rev: v2.17.4
+  rev: v3.2.6
   hooks:
   - id: pylint

.pylintrc

@@ -10,6 +10,7 @@ disable=
   duplicate-code,
   fixme,
   import-outside-toplevel,
+  invalid-field-call,
   invalid-name,
   missing-docstring,
   too-few-public-methods,

karapace/backup/api.py

@@ -22,6 +22,7 @@
 from .poll_timeout import PollTimeout
 from .topic_configurations import ConfigSource, get_topic_configurations
 from aiokafka.errors import KafkaError, TopicAlreadyExistsError
+from collections.abc import Sized
 from concurrent.futures import Future
 from confluent_kafka import Message, TopicPartition
 from enum import Enum
@@ -41,7 +42,7 @@
 from pathlib import Path
 from rich.console import Console
 from tenacity import retry, retry_if_exception_type, RetryCallState, stop_after_delay, wait_fixed
-from typing import Callable, Iterator, Literal, Mapping, NewType, Sized, TypeVar
+from typing import Callable, Iterator, Literal, Mapping, NewType, TypeVar
 
 import contextlib
 import datetime

karapace/backup/backends/v2.py

@@ -33,7 +33,10 @@ def safe_writer(
     ) -> Generator[IO[str], None, None]:
         with super().safe_writer(target, allow_overwrite) as buffer:
             buffer.write(V2_MARKER.decode())
-            yield buffer
+            try:
+                yield buffer
+            finally:
+                pass
 
 
 class SchemaBackupV2Writer(_BaseV2Writer):

karapace/backup/safe_writer.py

@@ -109,9 +109,12 @@ def str_writer(
     safe_context = _safe_temporary_descriptor(target.absolute(), allow_overwrite)
 
     with safe_context as fd, open(fd, "w") as buffer:
-        yield buffer
-        buffer.flush()
-        os.fsync(fd)
+        try:
+            yield buffer
+            buffer.flush()
+            os.fsync(fd)
+        finally:
+            pass
 
 
 def _check_destination_directory(destination: Path) -> None:

karapace/protobuf/known_dependency.py

@@ -16,7 +16,7 @@ def static_init(cls: Any) -> object:
     return cls
 
 
-@static_init
+@static_init  # pylint: disable=used-before-assignment
 class KnownDependency:
     index: Dict = dict()
     index_simple: Dict = dict()

karapace/protobuf/message_element.py

@@ -91,7 +91,7 @@ def to_schema(self) -> str:
         return "".join(result)
 
     def compare(self, other: TypeElement, result: CompareResult, types: CompareTypes) -> None:
-        from karapace.protobuf.compare_type_lists import compare_type_lists
+        from karapace.protobuf.compare_type_lists import compare_type_lists  # pylint: disable=cyclic-import
 
         if not isinstance(other, MessageElement):
             result.add_modification(Modification.TYPE_ALTER)

karapace/protobuf/proto_type.py

@@ -19,7 +19,7 @@ def static_init(cls) -> object:
     return cls
 
 
-@static_init
+@static_init  # pylint: disable=used-before-assignment
 class ProtoType:
     @property
     def simple_name(self) -> str:

karapace/protobuf/syntax_reader.py

@@ -103,7 +103,6 @@ def read_quoted_string(self) -> str:
                 self.newline()
 
         self.unexpected("unterminated string")
-        return ""
 
     def read_numeric_escape(self, radix: int, length: int) -> str:
         value = -1
@@ -180,7 +179,7 @@ def read_word(self) -> str:
         self.expect(start < self.pos, "expected a word")
         return self.data[start : self.pos]
 
-    def read_int(self) -> int:  # pylint: disable=inconsistent-return-statements
+    def read_int(self) -> int:
         """Reads an integer and returns it."""
         tag: str = self.read_word()
         try:

karapace/schema_reader.py

@@ -488,7 +488,7 @@ def _handle_msg_schema_hard_delete(self, key: dict) -> None:
         if self.database.find_subject(subject=subject) is None:
             LOG.warning("Hard delete: Subject %s did not exist, should have", subject)
         elif version not in self.database.find_subject_schemas(subject=subject, include_deleted=True):
-            LOG.warning("Hard delete: Version %d for subject %s did not exist, should have", version, subject)
+            LOG.warning("Hard delete: version: %r for subject: %r did not exist, should have", version, subject)
         else:
             LOG.info("Hard delete: subject: %r version: %r", subject, version)
             self.database.delete_subject_schema(subject=subject, version=version)

karapace/schema_registry.py

@@ -179,6 +179,8 @@ async def subject_delete_local(self, subject: Subject, permanent: bool) -> list[
                     version_list = list(schema_versions_live)
                     if version_list:
                         latest_version_id = version_list[-1]
+                    else:
+                        return []
 
                 referenced_by = self.schema_reader.get_referenced_by(subject, latest_version_id)
                 if referenced_by and len(referenced_by) > 0:

karapace/utils.py

@@ -203,7 +203,7 @@ def convert_to_int(object_: dict, key: str, content_type: str) -> None:
     try:
         object_[key] = int(object_[key])
     except ValueError:
-        from karapace.rapu import http_error
+        from karapace.rapu import http_error  # pylint: disable=cyclic-import
 
         http_error(
             message=f"{key} is not a valid int: {object_[key]}",

tests/unit/test_schema_reader.py

@@ -5,6 +5,7 @@
 See LICENSE for details
 """
 
+from _pytest.logging import LogCaptureFixture
 from concurrent.futures import ThreadPoolExecutor
 from confluent_kafka import Message
 from dataclasses import dataclass
@@ -20,12 +21,13 @@
     OFFSET_EMPTY,
     OFFSET_UNINITIALIZED,
 )
-from karapace.typing import SchemaId
+from karapace.typing import SchemaId, Version
 from tests.base_testcase import BaseTestCase
 from unittest.mock import Mock
 
 import confluent_kafka
 import json
+import logging
 import pytest
 import random
 import time
@@ -292,3 +294,27 @@ def test_soft_deleted_schema_storing() -> None:
 
     soft_deleted_stored_schema = schema_reader.database.find_schema(schema_id=SchemaId(1))
     assert soft_deleted_stored_schema is not None
+
+
+def test_handle_msg_delete_subject_logs(caplog: LogCaptureFixture) -> None:
+    database_mock = Mock(spec=InMemoryDatabase)
+    database_mock.find_subject.return_value = True
+    database_mock.find_subject_schemas.return_value = {
+        Version(1): "SchemaVersion"
+    }  # `SchemaVersion` is an actual object, simplified for test
+    schema_reader = KafkaSchemaReader(
+        config=DEFAULTS,
+        offset_watcher=OffsetWatcher(),
+        key_formatter=KeyFormatter(),
+        master_coordinator=None,
+        database=database_mock,
+    )
+
+    with caplog.at_level(logging.WARNING, logger="karapace.schema_reader"):
+        schema_reader._handle_msg_schema_hard_delete(  # pylint: disable=protected-access
+            key={"subject": "test-subject", "version": 2}
+        )
+        for log in caplog.records:
+            assert log.name == "karapace.schema_reader"
+            assert log.levelname == "WARNING"
+            assert log.message == "Hard delete: version: Version(2) for subject: 'test-subject' did not exist, should have"