confluentinc / confluentinc/confluent-kafka-python
SchemaRegistryClient caches incorrect entries ( or does not invalidate ) when using the `get_schema()` and `get_latest_version()` APIs.
- Dominant language
- Python
- Stars
- 509
- Forks
- 964
- Avg merge
- 2d 2h
- Merged PRs (30d)
- 14
Description
## Description :
There seems to be a bug in how the SchemaRegistryClient caches the subject entries ( with **[confluent_kafka v2.8.0](https://docs.confluent.io/platform/current/clients/confluent-kafka-python/html/_modules/confluent_kafka/schema_registry/schema_registry_client.html#SchemaRegistryClient.get_schema)** ) in the `SchemaRegistryClient.get_schema()` method. It can add a `subject_name=None` entry in the cache when only a schema_id is passed, and this does not get invalidated when the subject is permanently deleted off.
Adding details on how to reproduce this issue :
## Steps to reproduce :
- Created a SchemaRegistryClient instance and inspected the empty cache
```
In [1]: print(client._cache.schema_id_index); print(client._cache.schema_index)
defaultdict(, {})
defaultdict(, {})
```
- Registered a schema and the cache correctly reflects the subject-name entry mappings
```
In [3]: client.register_schema(subject_name, Schema(schema_str, 'AVRO'))
Out[3]: 8636
In [4]: print(client._cache.schema_id_index); print(client._cache.schema_index)
defaultdict(, {'testkafka.test_librdkafka_v2.8.0.schema_registry_issue_tests.on_version_2.8.0.release_notes_specific.test_schema_deletion_apis': {8636: Schema(schema_str=None, schema_type='AVRO', references=[], metadata=None, rule_set=None)}})
defaultdict(, {'testkafka.test_librdkafka_v2.8.0.schema_registry_issue_tests.on_version_2.8.0.release_notes_specific.test_schema_deletion_apis': {Schema(schema_str=None, schema_type='AVRO', references=[], metadata=None, rule_set=None): 8636}})
```
- Queried the schema with schema-id. And, now the updated cache seems to have a **None entry** ( in place of the subject-name ) that was added by `get_schema()`.
```
In [5]: client.get_schema(8636)
Out[5]: Schema(schema_str='{"type":"record","name":"value_record","fields":[{"name":"id","type":"int"},{"name":"name","type":"string"},{"name":"age","type":"int"},{"name":"money","type":"float"}]}', schema_type='AVRO', references=[], metadata=None, rule_set=None)
In [6]: print(client._cache.schema_id_index); print(client._cache.schema_index)
defaultdict(, {'testkafka.test_librdkafka_v2.8.0.schema_registry_issue_tests.on_version_2.8.0.release_notes_specific.test_schema_deletion_apis': {8636: Schema(schema_str=None, schema_type='AVRO', references=[], metadata=None, rule_set=None)}, None: {8636: Schema(schema_str='{"type":"record","name":"value_record","fields":[{"name":"id","type":"int"},{"name":"name","type":"string"},{"name":"age","type":"int"},{"name":"money","type":"float"}]}', schema_type='AVRO', references=[], metadata=None, rule_set=None)}})
defaultdict(, {'testkafka.test_librdkafka_v2.8.0.schema_registry_issue_tests.on_version_2.8.0.release_notes_specific.test_schema_deletion_apis': {Schema(schema_str=None, schema_type='AVRO', references=[], metadata=None, rule_set=None): 8636}, None: {Schema(schema_str='{"type":"record","name":"value_record","fields":[{"name":"id","type":"int"},{"name":"name","type":"string"},{"name":"age","type":"int"},{"name":"money","type":"float"}]}', schema_type='AVRO', references=[], metadata=None, rule_set=None): 8636}})
```
- On subject permanent deletion, now the client correctly deletes off the entry for the subject-name but not for the **None entries**.
```
In [7]: client.delete_subject(subject_name, permanent=False)
Out[7]: [1]
In [8]: client.delete_subject(subject_name, permanent=True)
Out[8]: [1]
In [9]: print(client._cache.schema_id_index); print(client._cache.schema_index)
defaultdict(, {None: {8636: Schema(schema_str='{"type":"record","name":"value_record","fields":[{"name":"id","type":"int"},{"name":"name","type":"string"},{"name":"age","type":"int"},{"name":"money","type":"float"}]}', schema_type='AVRO', references=[], metadata=None, rule_set=None)}})
defaultdict(, {None: {Schema(schema_str='{"type":"record","name":"value_record","fields":[{"name":"id","type":"int"},{"name":"name","type":"string"},{"name":"age","type":"int"},{"name":"money","type":"float"}]}', schema_type='AVRO', references=[], metadata=None, rule_set=None): 8636}})
```
- On further lookups ( with `get_schema()` ), we would now be reading from the cached **None entry** ( when a subject_name is not passed ). **Note** that this was the only subject for this schema-id, and on subject deletion the schema should also get deleted now.
```
In [10]: client.get_schema(8636)
Out[10]: Schema(schema_str='{"type":"record","name":"value_record","fields":[{"name":"id","type":"int"},{"name":"name","type":"string"},{"name":"age","type":"int"},{"name":"money","type":"float"}]}', schema_type='AVRO', references=[], metadata=None, rule_set=None)
```
However the schema and schema-id were actually deleted off ( verified with making a normal rest call ) -
```
{'error_code': 40403,
'message': 'Schema 8636 not found io.confluent.rest.exceptions.RestNotFoundException: Schema 8636 not found\nio.confluent.rest.exceptions.RestNotFoundException: Schema 8636 not found\n\tat io.confluent.kafka.schemaregistry.rest.exceptions.Errors.schemaNotFoundException(Errors.java:129)\n\tat io.confluent.kafka.schemaregistry.rest.resources.SchemasResource.getSchema(SchemasResource.java:168)\n\tat jdk.internal.reflect.GeneratedMethodAccessor14.invoke(Unknown Source)\n
...
```
## Impact :
As the cache **None entry** can never get invalidated / deleted off, then this means that there will always be a dangling reference to an update/deleted schema ( with an active SchemaRegistryClient instance ). And we can fetch this incorrect cache reference when using `get_schema(schema_id)` ( without the subject_name ).
---
### Caching issue with `get_latest_version()`
Also noting a behavior with the `get_latest_version()` method when using the **LRUCache** for the `_latest_version_cache` :
We are storing the subject-name as the cache-key when `get_latest_version()` is called. And this means that on a new schema version register, we never invalidate or update this old reference ( to version 1 ) and `get_latest_version()` would return this old value from cache.
```
In [2]: schema_str
Out[2]: '{"type":"record","name":"value_record","fields":[{"name":"id","type":"int"},{"name":"name","type":"string"},{"name":"age","type":"int"},{"name":"money","type":"float"}]}'
In [3]: client.register_schema(subject_name, Schema(schema_str, 'AVRO'))
Out[3]: 8632
In [4]: client.get_latest_version(subject_name)
Out[4]: RegisteredSchema(schema_id=8632, schema=Schema(schema_str='{"type":"record","name":"value_record","fields":[{"name":"id","type":"int"},{"name":"name","type":"string"},{"name":"age","type":"int"},{"name":"money","type":"float"}]}', schema_type='AVRO', references=[], metadata=None, rule_set=None), subject='testkafka.test_librdkafka_v2.8.0.schema_registry_issue_tests.on_version_2.8.0.release_notes_specific.test_schema_deletion_apis', version=1)
In [6]: schema_str
Out[6]: '{"type":"record","name":"value_record","fields":[{"name":"id","type":"int"},{"name":"name","type":"string"},{"name":"age","type":"int"},{"name":"money","type":"float"},{"name":"hobby","type":["null","string"],"default":"null"}]}'
In [7]: client.register_schema(subject_name, Schema(schema_str, 'AVRO'))
Out[7]: 8633
# This give incorrect output
In [8]: client.get_latest_version(subject_name)
Out[8]: RegisteredSchema(schema_id=8632, schema=Schema(schema_str='{"type":"record","name":"value_record","fields":[{"name":"id","type":"int"},{"name":"name","type":"string"},{"name":"age","type":"int"},{"name":"money","type":"float"}]}', schema_type='AVRO', references=[], metadata=None, rule_set=None), subject='testkafka.test_librdkafka_v2.8.0.schema_registry_issue_tests.on_version_2.8.0.release_notes_specific.test_schema_deletion_apis', version=1)
```
Could we have a look at the above 2 bugs in caching behavior, and if we can make the SchemaRegistryClient cache more robust ?
Contributor guide
No contributing guide indexed for this repository
Assessment
This issue has not been assessed yet.