Aiven-Open / Aiven-Open/karapace

make the creation of Subjects explicit in topic producing

未關閉
#721 0 則留言 0 個 reaction 已指派 0 人 在 GitHub 檢視
主要語言
Python
星號
634
分支
110
平均合併
4 天 7 小時
30 天內合併 PR
4

描述

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

貢獻指南

開啟貢獻指南

評估

這個 Issue 還沒有評估資料。

把新 issue 寄到你的電子郵件信箱

精選適合新手參與的 GitHub issue 摘要。