Fix name_strategy configuration option

Allow setting `name_strategy` from config, and fix documentation to
match default that has been in the implementation so far.

README.rst

@@ -460,7 +460,7 @@ Keys to take special care are the ones needed to configure Kafka and advertised_
      - ``runtime``
      - Runtime directory for the ``protoc`` protobuf schema parser and code generator
    * - ``name_strategy``
-     - ``subject_name``
+     - ``topic_name``
      - Name strategy to use when storing schemas from the kafka rest proxy service
    * - ``master_election_strategy``
      - ``lowest``

karapace/config.py

@@ -73,6 +73,7 @@ class Config(TypedDict):
     session_timeout_ms: int
     karapace_rest: bool
     karapace_registry: bool
+    name_strategy: str
     master_election_strategy: str
     protobuf_runtime_directory: str
 
@@ -142,6 +143,7 @@ class ConfigDefaults(Config, total=False):
     "session_timeout_ms": 10000,
     "karapace_rest": False,
     "karapace_registry": False,
+    "name_strategy": "topic_name",
     "master_election_strategy": "lowest",
     "protobuf_runtime_directory": "runtime",
 }
@@ -158,6 +160,13 @@ class ElectionStrategy(Enum):
     lowest = "lowest"
 
 
+@unique
+class NameStrategy(Enum):
+    topic_name = "topic_name"
+    record_name = "record_name"
+    topic_record_name = "topic_record_name"
+
+
 def parse_env_value(value: str) -> str | int | bool:
     # we only have ints, strings and bools in the config
     try:
@@ -256,6 +265,13 @@ def validate_config(config: Config) -> None:
             f"Invalid master election strategy: {master_election_strategy}, valid values are {valid_strategies}"
         ) from None
 
+    name_strategy = config["name_strategy"]
+    try:
+        NameStrategy(name_strategy)
+    except ValueError:
+        valid_strategies = [strategy.value for strategy in NameStrategy]
+        raise InvalidConfiguration(f"Invalid name strategy: {name_strategy}, valid values are {valid_strategies}") from None
+
     if config["rest_authorization"] and config["sasl_bootstrap_uri"] is None:
         raise InvalidConfiguration(
             "Using 'rest_authorization' requires configuration value for 'sasl_bootstrap_uri' to be set"

karapace/serialization.py

@@ -229,8 +229,6 @@ class SchemaRegistrySerializer:
     def __init__(
         self,
         config: dict,
-        name_strategy: str = "topic_name",
-        **cfg,  # pylint: disable=unused-argument
     ) -> None:
         self.config = config
         self.state_lock = asyncio.Lock()
@@ -245,7 +243,8 @@ def __init__(
         else:
             registry_url = f"http://{self.config['registry_host']}:{self.config['registry_port']}"
             registry_client = SchemaRegistryClient(registry_url, session_auth=session_auth)
-        self.subject_name_strategy = NAME_STRATEGIES[name_strategy]
+        name_strategy = config.get("name_strategy", "topic_name")
+        self.subject_name_strategy = NAME_STRATEGIES.get(name_strategy, topic_name_strategy)
         self.registry_client: Optional[SchemaRegistryClient] = registry_client
         self.ids_to_schemas: Dict[int, TypedSchema] = {}
         self.ids_to_subjects: MutableMapping[int, List[Subject]] = TTLCache(maxsize=10000, ttl=600)

tests/unit/test_protobuf_serialization.py

@@ -29,7 +29,7 @@
 async def make_ser_deser(config_path: str, mock_client) -> SchemaRegistrySerializer:
     with open(config_path, encoding="utf8") as handler:
         config = read_config(handler)
-    serializer = SchemaRegistrySerializer(config_path=config_path, config=config)
+    serializer = SchemaRegistrySerializer(config=config)
     await serializer.registry_client.close()
     serializer.registry_client = mock_client
     return serializer

tests/unit/test_serialization.py

@@ -33,7 +33,7 @@
 async def make_ser_deser(config_path: str, mock_client) -> SchemaRegistrySerializer:
     with open(config_path, encoding="utf8") as handler:
         config = read_config(handler)
-    serializer = SchemaRegistrySerializer(config_path=config_path, config=config)
+    serializer = SchemaRegistrySerializer(config=config)
     await serializer.registry_client.close()
     serializer.registry_client = mock_client
     return serializer