Ensure the deletion consistency of topic and schema - #14608
Conversation
| futures.add(deleteSchema().thenApply(schemaVersion -> null)); | ||
| } | ||
|
|
||
| futures.add(deleteSchema().thenApply(schemaVersion -> null)); |
There was a problem hiding this comment.
What happens to topics with multiple versions fo schema? Will all versions be cleaned?
There was a problem hiding this comment.
Yes, will delete all the versions
There was a problem hiding this comment.
Hmmm, it's a little confusing for using persistent schema storage for a non-persistent topic, after the broker restart and the producer/consumer will not connect to a non-persistent topic, the non-persistent topic is equivalent to being deleted, In some cases that users using a random name for non-persistent topic, after the application restarts, the topic name will be changes, looks like here will introduce a schema resource leak.
codelipenghui
left a comment
There was a problem hiding this comment.
I noticed there is a behavior change that might impact users' applications if the schema auto uploading is disabled.
Before this change, users can delete the topic without deleting schema, so that the application can connect to the newly created topic with the same topic name and existing schema, but after this change, not able to connect to the topic again.
Hmmm, before, users usually use the delete topic to truncate data of the topic. After #10326 introduced truncate topic, I think for now users should to use the truncate method to cleanup data of the topic. So there is no reason why the user still wants to use this topic, but also delete it.
We'd better clarify in the document.
| futures.add(deleteSchema().thenApply(schemaVersion -> null)); | ||
| } | ||
|
|
||
| futures.add(deleteSchema().thenApply(schemaVersion -> null)); |
There was a problem hiding this comment.
Yes, will delete all the versions
| futures.add(deleteSchema().thenApply(schemaVersion -> null)); | ||
| } | ||
|
|
||
| futures.add(deleteSchema().thenApply(schemaVersion -> null)); |
There was a problem hiding this comment.
Hmmm, it's a little confusing for using persistent schema storage for a non-persistent topic, after the broker restart and the producer/consumer will not connect to a non-persistent topic, the non-persistent topic is equivalent to being deleted, In some cases that users using a random name for non-persistent topic, after the application restarts, the topic name will be changes, looks like here will introduce a schema resource leak.
I have added explanation in the CLI / API. |
codelipenghui
left a comment
There was a problem hiding this comment.
Please also help add a test to make sure the schema been removed after the topic has been deleted.
| "--deleteSchema" }, description = "Delete schema while deleting topic") | ||
| @Parameter(names = {"-d", "--deleteSchema"}, description = "Delete schema while deleting topic, " | ||
| + "but the parameter is invalid and the schema is always deleted", hidden = true) | ||
| private boolean deleteSchema = false; |
There was a problem hiding this comment.
Should also update the default value to true?
There was a problem hiding this comment.
Maybe we can't change the default value to true because that would change its usage.
Before modification: it means false if you don't specify -d, and it means true if you specify -d;
After modification: it means true if you don’t specify -d, but it means false only you specify -d false.
The two are not fully compatible.
It is similar to #12128
| "--deleteSchema" }, description = "Delete schema while deleting topic") | ||
| @Parameter(names = {"-d", "--deleteSchema"}, description = "Delete schema while deleting topic, " | ||
| + "but the parameter is invalid and the schema is always deleted", hidden = true) | ||
| private boolean deleteSchema = false; |
There was a problem hiding this comment.
Should also update the default value to true?
| if (deleteSchema) { | ||
| return pulsar().getBrokerService() | ||
| .deleteSchemaStorage(topicName.getPartition(0).toString()) | ||
| .thenCompose(unused -> | ||
| internalRemovePartitionsAuthenticationPoliciesAsync(numPartitions)) | ||
| .thenCompose(unused2 -> | ||
| internalRemovePartitionsTopicAsync(numPartitions, force)); | ||
| } |
There was a problem hiding this comment.
If remove these lines, how does the method remove the schema?
There was a problem hiding this comment.
internalRemovePartitionsTopicAsync(numPartitions, force) will delete schema by default.
| return t.deleteSchema().thenCompose(schemaVersion -> { | ||
| log.info("Successfully delete topic {}'s schema of version {}", t.getName(), schemaVersion); | ||
| return t.delete(); | ||
| }); |
There was a problem hiding this comment.
If remove these lines how do we delete the schema?
There was a problem hiding this comment.
t.delete() will delete schema by default.
details: delete() -> delete(boolean,boolean,boolean) -> deleteSchema()
The test testDeleteTopicAndSchemaForV1 in |
* Ensure the deletion consistency of topic and schema (cherry picked from commit c06b17d)
* Ensure the deletion consistency of topic and schema (cherry picked from commit c06b17d)
Master Issue: #12795
Dev email: discuss and reply
Motivation
Deleting topic also deletes its schema.
Documentation
no-need-doc