Fix messages in TopicPolicies will never be cleaned up #11928
Conversation
315157973
requested review from
eolivelli,
codelipenghui,
hangc0276 and
gaoran10
|
@gaoran10 PTAL, it seems that the transaction also has this problem |
...ker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
Outdated
Show resolved
Hide resolved
| @@ -330,9 +343,6 @@ private void refreshTopicPoliciesCache(Message<PulsarEvent> msg) { | |||
| case UPDATE: | |||
| policiesCache.put(topicName, event.getPolicies()); | |||
| break; | |||
| case DELETE: | |||
There was a problem hiding this comment.
what about old data present in existing clusters ?
There was a problem hiding this comment.
Good catch, maybe we could call the delete method to publish a message with a null value in branch DELETE?
There was a problem hiding this comment.
Indeed, for the compatibility, this code cannot be deleted now, I will add a comment here.
But Topic's existing old data seems to need to provide additional tools to delete it.
There was a problem hiding this comment.
When recovering the topic policies, the reader will read messages from the earliest position, it could read the DELETE type Pulsar event, does it handle the DELETE event messages at this time?
There was a problem hiding this comment.
I added a compatibility adaptation
| SystemTopicClient<PulsarEvent> systemTopicClient = namespaceEventsSystemTopicFactory | ||
| .createTopicPoliciesSystemTopicClient(topicName.getNamespaceObject()); | ||
| systemTopicClient.newWriterAsync().thenAccept(writer | ||
| -> writer.deleteAsync(getPulsarEvent(topicName, ActionType.DELETE, null))); |
There was a problem hiding this comment.
The writer should be closed and for replay the historical policies, we should use one writer instead create writer for each policy data?
There was a problem hiding this comment.
I have closed the Writer.
Now all Writers in SystemTopicBasedTopicPoliciesService seem to be created every time.
I think this is reasonable, after all, Topic Policies are not high-frequency operations.
|
@315157973 Thanks for your contribution. For this PR, do we need to update docs? (The PR template contains info about doc, which helps others know more about the changes. Can you provide doc-related info in this and future PR descriptions? Thanks) |
|
Hi, @Anonymitaet, I think this doesn't need a doc, because this is a bug fix. |
(cherry picked from commit 93e2db0)
Motivation
The current Topic Policies use compaction topic.
When a compaction topic deletes a message, the body of the message must be null.
Now, even if policies are being deleted, the body of the message is not null.
This will cause some messages in compaction topic never be cleaned up.
Modifications
1）When deleting policies, set the message body to null
2）Add a delete API instead of modifying write API