Fix SR delete behavior with client-side caching

Author: fangnx

src/confluent_kafka/schema_registry/_async/schema_registry_client.py

@@ -1174,12 +1174,14 @@ async def delete_version(self, subject_name: str, version: int, permanent: bool
             response = await self._rest_client.delete(
                 'subjects/{}/versions/{}?permanent=true'.format(_urlencode(subject_name), version)
             )
-            self._cache.remove_by_subject_version(subject_name, version)
         else:
             response = await self._rest_client.delete(
                 'subjects/{}/versions/{}'.format(_urlencode(subject_name), version)
             )
 
+        # Clear cache for both soft and hard deletes to maintain consistency
+        self._cache.remove_by_subject_version(subject_name, version)
+
         return response
 
     async def set_compatibility(self, subject_name: Optional[str] = None, level: Optional[str] = None) -> str:

src/confluent_kafka/schema_registry/_sync/schema_registry_client.py

@@ -1174,12 +1174,14 @@ def delete_version(self, subject_name: str, version: int, permanent: bool = Fals
             response = self._rest_client.delete(
                 'subjects/{}/versions/{}?permanent=true'.format(_urlencode(subject_name), version)
             )
-            self._cache.remove_by_subject_version(subject_name, version)
         else:
             response = self._rest_client.delete(
                 'subjects/{}/versions/{}'.format(_urlencode(subject_name), version)
             )
 
+        # Clear cache for both soft and hard deletes to maintain consistency
+        self._cache.remove_by_subject_version(subject_name, version)
+
         return response
 
     def set_compatibility(self, subject_name: Optional[str] = None, level: Optional[str] = None) -> str:

src/confluent_kafka/schema_registry/common/schema_registry_client.py

@@ -268,11 +268,12 @@ def remove_by_subject_version(self, subject: str, version: int):
 
         with self.lock:
             if subject in self.rs_id_index:
-                for schema_id, registered_schema in self.rs_id_index[subject].items():
+                for schema_id, registered_schema in list(self.rs_id_index[subject].items()):
                     if registered_schema.version == version:
-                        del self.rs_schema_index[subject][schema_id]
+                        del self.rs_id_index[subject][schema_id]
+
             if subject in self.rs_schema_index:
-                for schema, registered_schema in self.rs_schema_index[subject].items():
+                for schema, registered_schema in list(self.rs_schema_index[subject].items()):
                     if registered_schema.version == version:
                         del self.rs_schema_index[subject][schema]
             rs = None
@@ -284,6 +285,7 @@ def remove_by_subject_version(self, subject: str, version: int):
                 if subject in self.schema_id_index:
                     if rs.schema_id in self.schema_id_index[subject]:
                         del self.schema_id_index[subject][rs.schema_id]
+                if subject in self.schema_index:
                     if rs.schema in self.schema_index[subject]:
                         del self.schema_index[subject][rs.schema]
 

tests/integration/schema_registry/_async/test_api_client.py

@@ -422,6 +422,34 @@ async def test_api_delete_subject_version(kafka_cluster, load_file):
     assert subject not in await sr.get_subjects()
 
 
+async def test_api_delete_version_soft_then_hard(kafka_cluster, load_file):
+    """
+    Performs a soft delete followed by a hard delete with cache populated and cleared correctly.
+    """
+    sr = kafka_cluster.async_schema_registry()
+
+    schema = Schema(load_file('basic_schema.avsc'), schema_type='AVRO')
+    subject = str(uuid1())
+
+    # Register schema and trigger cache population
+    await sr.register_schema(subject, schema)
+    registered = await sr.lookup_schema(subject, schema)
+    version = registered.version
+    assert sr._cache.get_registered_by_subject_version(subject, version) is not None
+
+    # Verify soft delete clears cache
+    deleted_version = await sr.delete_version(subject, version, permanent=False)
+    assert deleted_version == version
+    assert sr._cache.get_registered_by_subject_version(subject, version) is None
+
+    # Verify hard delete proceeds without error
+    deleted_version = await sr.delete_version(subject, version, permanent=True)
+    assert deleted_version == version
+
+    # Verify subject is fully deleted
+    assert subject not in await sr.get_subjects()
+
+
 async def test_api_subject_config_update(kafka_cluster, load_file):
     """
     Updates a subjects compatibility policy then ensures the same policy

tests/integration/schema_registry/_sync/test_api_client.py

@@ -422,6 +422,34 @@ def test_api_delete_subject_version(kafka_cluster, load_file):
     assert subject not in sr.get_subjects()
 
 
+def test_api_delete_version_soft_then_hard(kafka_cluster, load_file):
+    """
+    Performs a soft delete followed by a hard delete with cache populated and cleared correctly.
+    """
+    sr = kafka_cluster.schema_registry()
+
+    schema = Schema(load_file('basic_schema.avsc'), schema_type='AVRO')
+    subject = str(uuid1())
+
+    # Register schema and trigger cache population
+    sr.register_schema(subject, schema)
+    registered = sr.lookup_schema(subject, schema)
+    version = registered.version
+    assert sr._cache.get_registered_by_subject_version(subject, version) is not None
+
+    # Verify soft delete clears cache
+    deleted_version = sr.delete_version(subject, version, permanent=False)
+    assert deleted_version == version
+    assert sr._cache.get_registered_by_subject_version(subject, version) is None
+
+    # Verify hard delete proceeds without error
+    deleted_version = sr.delete_version(subject, version, permanent=True)
+    assert deleted_version == version
+
+    # Verify subject is fully deleted
+    assert subject not in sr.get_subjects()
+
+
 def test_api_subject_config_update(kafka_cluster, load_file):
     """
     Updates a subjects compatibility policy then ensures the same policy