Aiven-Open / Aiven-Open/karapace

make the creation of Subjects explicit in topic producing

Aperta
#721 0 commenti 0 reazioni 0 assegnatari Vedi su GitHub
Lingua principale
Python
Stelle
634
Fork
110
Merge medio
4g 7h
PR unite (30g)
4

Descrizione

Currently if we do a POST request towards `/topics/topic_name` with only the `key_schema`, `value_schema` and the `records` in the body of the request we implicitly create a Subject associated to the topic.
I think doing this implicitly without a flag is dangerous and can lead users to create new Subject even if they don't realize it.
It also creates different behaviour in the request based on the absence/presence of fields.
An example of that was related to the issue #720.

If we try to produce by specifying the id's of the key and the value we get back an error because the subject do not have the `-key` and the `-value` postfix.

Otherwise, if we get rid of the ids in the request and we include the schema, the schema registry proceed automatically with creating the new subjects giving the false impression that the previous registration of the Subject was responsible for the message produced.

Code to replicate the issue:
```python3
async def test_another_avro_publish(
rest_async_client: Client,
registry_async_client: Client,
admin_client: KafkaRestAdminClient,
):
topic = new_topic(admin_client)
other_tn = new_topic(admin_client)

await wait_for_topics(rest_async_client, topic_names=[topic, other_tn], timeout=NEW_TOPIC_TIMEOUT, sleep=1)
header = REST_HEADERS["avro"]

tested_avro_schema = {
"type": "record",
"name": "example",
"namespace": "example",
"doc": "example",
"fields": [
{
"type": "int",
"name": "test",
"doc": "my test number",
"namespace": "test",
"default": "5"
}
]
}

schema_str = json.dumps(tested_avro_schema)

res = await registry_async_client.post(f"subjects/{topic}/versions", json={
"schema": schema_str,
"schemaType": "AVRO"
})
assert res.ok

key_schema_id = value_schema_id = res.json()["id"]
key_body = {
"test": 5
}

value_body = {
"test": 5
}

body = {
"key_schema": schema_str,
"value_schema": schema_str,
"records": [
{
"key": key_body,
"value": value_body
}
]
}

url = f"/topics/{topic}"
res = await rest_async_client.post(url, json=body, headers=header)
assert res.ok
```

This works fine (by implicitly creating two new subjects), instead if we keep the same code but we replace the body with the following one, it give us the error that the subjects aren't registered (this works the same even if we get rid of the `key_schema` and the `value_schema`):
```python3
body = {
"key_schema_id": key_schema_id,
"key_schema": schema_str,
"value_schema_id": value_schema_id, # topic -> subject -> schema
"value_schema": schema_str,
"records": [
{
"key": key_body,
"value": value_body
}
]
}
```

The correct way of registering the subjects and deal with it it's the following:
```python3
async def test_another_avro_publish(
rest_async_client: Client,
registry_async_client: Client,
admin_client: KafkaRestAdminClient,
):
topic = new_topic(admin_client)
other_tn = new_topic(admin_client)

await wait_for_topics(rest_async_client, topic_names=[topic, other_tn], timeout=NEW_TOPIC_TIMEOUT, sleep=1)
header = REST_HEADERS["avro"]

tested_avro_schema = {
"type": "record",
"name": "example",
"namespace": "example",
"doc": "example",
"fields": [
{
"type": "int",
"name": "test",
"doc": "my test number",
"namespace": "test",
"default": "5"
}
]
}

schema_str = json.dumps(tested_avro_schema)

# check succeeds with 1 record and brand new schema]
res = await registry_async_client.post(f"subjects/{topic}-key/versions", json={
"schema": schema_str,
"schemaType": "AVRO"
})
assert res.ok

key_schema_id = res.json()["id"]

res = await registry_async_client.post(f"subjects/{topic}-value/versions", json={
"schema": schema_str,
"schemaType": "AVRO"
})
assert res.ok

value_schema_id = res.json()["id"]

key_body = {
"test": 5
}

value_body = {
"test": 5
}

body = {
"key_schema_id": key_schema_id,
"value_schema_id": value_schema_id,
"records": [
{
"key": key_body,
"value": value_body
}
]
}

url = f"/topics/{topic}"
res = await rest_async_client.post(url, json=body, headers=header)
assert res.ok
```

That works fine in both ways, if we include the ids or if we include the schemas.

I suggest to unify the behaviour by explicitly adding a field when we want an upsert semantics for the Subjects

Guida per i contributori

Apri la guida per i contributori

Valutazione

Questa issue non è ancora stata valutata.

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.