[fix] [broker] Ensure policy cache is updated before updateTopicPoliciesAsync returns. by thetumbled · Pull Request #21030 · apache/pulsar

thetumbled · 2023-08-19T04:43:37Z

Motivation

Calling these two method synchronously to set the retention and ttl of one topic will result into unexpected results.

admin.topics().setRetention
admin.topics().setMessageTTL

For example, if we call

int retentionTimeInMinutes = 2880;
int messageTTLInSecond = retentionTimeInMinutes * 60;
admin.topics().setRetention(topic,new RetentionPolicies(retentionTimeInMinute, -1));
admin.topics().setMessageTTL(topic, messageTTLInSecond);

We expect that the retention is set to 2880 minutes, ttl is set to 2880*16=172800 seconds. But the result is that the retention policy is null !
We read the messages from __change_events derectly and find out the reason:

TimeStamp:1692157480789 ,Message received: PulsarEvent(eventType=TOPIC_POLICY, actionType=UPDATE, topicPoliciesEvent=TopicPoliciesEvent(, policies=TopicPolicies(backLogQuotaMap={}, subscriptionTypesEnabled=[], persistence=null, **retentionPolicies=RetentionPolicies{retentionTimeInMinutes=2880, retentionSizeInMB=-1},** deduplicationEnabled=null, **messageTTLInSeconds=null**, maxProducerPerTopic=null, maxConsumerPerTopic=null, maxConsumersPerSubscription=null, maxUnackedMessagesOnConsumer=null, maxUnackedMessagesOnSubscription=null, delayedDeliveryTickTimeMillis=null, delayedDeliveryEnabled=null, offloadPolicies=null, inactiveTopicPolicies=null, dispatchRate=null, subscriptionDispatchRate=null, compactionThreshold=null, publishRate=null, subscribeRate=null, deduplicationSnapshotIntervalSeconds=null, maxMessageSize=null, maxSubscriptionsPerTopic=null, replicatorDispatchRate=null)))

TimeStamp:1692157480811 ,Message received: PulsarEvent(eventType=TOPIC_POLICY, actionType=UPDATE, topicPoliciesEvent=TopicPoliciesEvent(, policies=TopicPolicies(backLogQuotaMap={}, subscriptionTypesEnabled=[], persistence=null, **retentionPolicies=null**, deduplicationEnabled=null, **messageTTLInSeconds=172800**, maxProducerPerTopic=null, maxConsumerPerTopic=null, maxConsumersPerSubscription=null, maxUnackedMessagesOnConsumer=null, maxUnackedMessagesOnSubscription=null, delayedDeliveryTickTimeMillis=null, delayedDeliveryEnabled=null, offloadPolicies=null, inactiveTopicPolicies=null, dispatchRate=null, subscriptionDispatchRate=null, compactionThreshold=null, publishRate=null, subscribeRate=null, deduplicationSnapshotIntervalSeconds=null, maxMessageSize=null, maxSubscriptionsPerTopic=null, replicatorDispatchRate=null)))

We can see that the retention policy in the second message is null, which result into the null retention policy.

Diving into the implementation of SystemTopicBasedTopicPoliciesService, we can see that org.apache.pulsar.broker.service.SystemTopicBasedTopicPoliciesService#updateTopicPoliciesAsync returns as soon as the message is written into __change_events, but do not guranteen that the topic policy in cache is updated. However, topic policy setting api will retrieve the topic policy from cache.

org.apache.pulsar.broker.admin.v2.PersistentTopics#setRetention
org.apache.pulsar.broker.admin.v2.PersistentTopics#setMessageTTL

Before the cache is updated, the second rest api for setting ttl arrived, and change the topic policy based on the old one, which result into the problem.

Modifications

The solution is: Ensure policy cache is updated before updateTopicPoliciesAsync returns.

Verifying this change

Make sure that the change passes the CI checks.

(Please pick either of the following options)

This change added tests and can be verified as follows:

Does this pull request potentially affect one of the following parts:

If the box was checked, please highlight the changes

Documentation

doc
doc-required
doc-not-needed
doc-complete

Matching PR in forked repository

PR in forked repository: thetumbled#25

…rns.

github-actions · 2023-08-19T04:44:02Z

@thetumbled Please add the following content to your PR description and select a checkbox:

- [ ] `doc` <!-- Your PR contains doc changes -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [ ] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->

AnonHxy · 2023-08-19T15:38:07Z

                .thenAccept(msg -> {
                    refreshTopicPoliciesCache(msg);
                    notifyListener(msg);
+                    completeResults(TopicName.get(TopicName.get(msg.getKey()).getPartitionedTopicName()));


I dont think this is a right way to solve the problem. For examle:

If we have three different actions, e.g., setA and setB and setC:

setA is waiting for futureA completed. The eventMsg is eventA.

setB is waiting for futureB completed. The eventMsg is eventB.

setC is waiting for futureC completed. The eventMsg is eventC.

The method completeResults will complete futureA and futureB and futureC when read eventA. Howerver setB has not been refresh to the policies cache.

So setC maybe read the old B value

Yes, this is the corner case for this solution, which can not handle with concurrent request for modifying topic policy. This solution fix the synchronize case.
And i wonder whether we have the need to implement strong concurrent controll? There are two directions:

add a warning to users, that the topic policy setting api can not handle with concurrent cases

or, i can implement a stronger concurrent controll.
What do you think? @AnonHxy @Demogorgon314 @codelipenghui

OK. I see. Synchronize case is enough I think

codelipenghui · 2023-08-21T15:27:57Z

    @VisibleForTesting
    final Map<TopicName, List<TopicPolicyListener<TopicPolicies>>> listeners = new ConcurrentHashMap<>();

+    final Map<TopicName, BlockingDeque<CompletableFuture>> results = new ConcurrentHashMap<>();


I don't think it will resolve the issue completely because the next request might happened on different brokers.

All request for setting topic policy will be redirected to the owner broker of the partition-0, so we do not need to consider distributed situations.

All request for setting topic policy will be redirected to the owner broker of the partition-0, so we do not need to consider distributed situations.

Could you share the code which routed the HTTP request to the "owner broker of the partition-0"? I want to double confirm it. Thanks

we have such redirection logic in: org.apache.pulsar.broker.admin.impl.PersistentTopicsBase#preValidation

return getPartitionedTopicMetadataAsync(topicName, false, false) .thenCompose(metadata -> { if (metadata.partitions > 0) { return validateTopicOwnershipAsync(TopicName.get(topicName.toString() + PARTITIONED_TOPIC_SUFFIX + 0), authoritative); } else { return validateTopicOwnershipAsync(topicName, authoritative); } });

There are two other concerns about the fix:

The problem still exists if the Topic Owner is transferred to another broker

Now the reading is correct. But the topic may have yet to apply the latest policy even if users received the response.

I'm also thinking about how we can implement this function simpler.

Indeed, even if all the read and update operations occur on the same broker, it remains crucial to serialize all such operations. Without serialization, the broker would still be incapable of ensuring that every modification is carried out on the most up-to-date data.

Update from v1 -> v2 Update from v1 -> v3

The changes from v2 will be missed.

Additionally, during the transfer of ownership, it is possible for ongoing updates to continue taking place on the old broker. Meanwhile, new requests directed to the new broker may still have an opportunity to update the data based on a potentially "dirty read."

mattisonchao · 2023-08-22T01:29:54Z

From my understanding now. It should be a problem like a dirty reading. Right?

Since the design of the topic policy mechanism is the final consensus, it's a little bit hard for us to fix this kinda situation.

IMO, we can try to define a redirect logic here.

non-partitioned topic policy redirect to owner broker.
partitioned topic policy redirects to the partition-0's owner broker as a logical topic leader.

Afterwards, we move a distributed problem to a single machine which will be easier for us to handle it.

Defining a sync method to let the reader read the end of the topic at that moment is better than saving the operation future everywhere from my point of view.

thetumbled · 2023-08-22T02:08:51Z

From my understanding now. It should be a problem like a dirty reading. Right?

Since the design of the topic policy mechanism is the final consensus, it's a little bit hard for us to fix this kinda situation.

IMO, we can try to define a redirect logic here.

non-partitioned topic policy redirect to owner broker.

partitioned topic policy redirects to the partition-0's owner broker as a logical topic leader.

Yes, it is dirty reading problem.
Actually, we have such redirection logic in: org.apache.pulsar.broker.admin.impl.PersistentTopicsBase#preValidation

                        return getPartitionedTopicMetadataAsync(topicName, false, false)
                            .thenCompose(metadata -> {
                                if (metadata.partitions > 0) {
                                    return validateTopicOwnershipAsync(TopicName.get(topicName.toString()
                                    + PARTITIONED_TOPIC_SUFFIX + 0), authoritative);
                                } else {
                                    return validateTopicOwnershipAsync(topicName, authoritative);
                                }
                        });

So we do not need to consider distributed situations. @codelipenghui

Defining a sync method to let the reader read the end of the topic at that moment is better than saving the operation future everywhere from my point of view.

Currently, we have two places need to read the messages, the first one is in the backgroup thread triggered by reader.readNextAsync() asynchronously, the second one is here.

To achieve strong concurrency control, maybe we could make the whole setting logic running synchronously in same thread for each topic, which will make break change. But as you say, the design of the topic policy mechanism is the final consensus, we do not need such kind of strong guarantee, right?

IMO, i don't see any drawback of this solution in synchronous requests situations, which is the basic situation and enough for users. In synchronous requests situations, there will be only one result in the queue at most at a time, which will be removed before the corresponding request returns.

github-actions · 2023-10-06T01:47:43Z

The pr had no activity for 30 days, mark with Stale label.

lhotari · 2025-08-13T06:25:07Z

The problem has been addressed with "PIP-428: Change TopicPoliciesService interface to fix consistency issues" and the implementation PR #24427.

One of the main differences compared to the approach in this PR is that existing TopicPolicies instances are never mutated. Instead, the existing instance is cloned and the mutation happens on the cloned instance. This ensures that threads that are referencing the instance won't get an inconsistent view on the instance. The mutation will be atomic from this perspective and become available via the cache. A similar pattern is used for namespace policy updates.

lhotari · 2025-08-27T06:23:24Z

Closing since this has been addressed by PIP-428 implementation PR #24427

Ensure that the cache is updated before updateTopicPoliciesAsync retu…

3b8fb93

…rns.

github-actions Bot added the doc-label-missing label Aug 19, 2023

github-actions Bot added doc-not-needed Your PR changes do not impact docs and removed doc-label-missing labels Aug 19, 2023

thetumbled mentioned this pull request Aug 19, 2023

Ensure that the cache is updated before updateTopicPoliciesAsync retu… thetumbled/pulsar#25

Open

AnonHxy reviewed Aug 19, 2023

View reviewed changes

thetumbled requested a review from AnonHxy August 21, 2023 02:07

remove the result blocking queue if topic is removed.

7a91103

codelipenghui requested review from BewareMyPower, Technoboy-, mattisonchao and poorbarcode August 21, 2023 07:16

codelipenghui assigned thetumbled Aug 21, 2023

codelipenghui added this to the 3.2.0 milestone Aug 21, 2023

codelipenghui added the type/bug The PR fixed a bug or issue reported a bug label Aug 21, 2023

AnonHxy reviewed Aug 21, 2023

View reviewed changes

Comment thread ...ker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java

add test code.

73393a6

thetumbled requested a review from AnonHxy August 21, 2023 08:23

AnonHxy approved these changes Aug 21, 2023

View reviewed changes

codelipenghui requested changes Aug 21, 2023

View reviewed changes

codelipenghui added the category/reliability The function does not work properly in certain specific environments or failures. e.g. data lost label Aug 21, 2023

thetumbled requested a review from codelipenghui August 22, 2023 02:22

github-actions Bot added the Stale label Oct 6, 2023

Technoboy- modified the milestones: 3.2.0, 3.3.0 Dec 22, 2023

coderzc removed this from the 3.3.0 milestone May 8, 2024

coderzc added this to the 3.4.0 milestone May 8, 2024

lhotari modified the milestones: 4.0.0, 4.1.0 Oct 11, 2024

lhotari closed this Aug 27, 2025

Uh oh!

Conversation

thetumbled commented Aug 19, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Motivation

Modifications

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

Matching PR in forked repository

Uh oh!

github-actions Bot commented Aug 19, 2023

Uh oh!

Uh oh!

AnonHxy Aug 19, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

thetumbled Aug 21, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

AnonHxy Aug 21, 2023

Choose a reason for hiding this comment

Uh oh!

Uh oh!

codelipenghui Aug 21, 2023

Choose a reason for hiding this comment

Uh oh!

thetumbled Aug 22, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

poorbarcode Aug 22, 2023

Choose a reason for hiding this comment

Uh oh!

thetumbled Aug 22, 2023

Choose a reason for hiding this comment

Uh oh!

poorbarcode Aug 23, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

codelipenghui Sep 5, 2023

Choose a reason for hiding this comment

Uh oh!

mattisonchao commented Aug 22, 2023

Uh oh!

thetumbled commented Aug 22, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

github-actions Bot commented Oct 6, 2023

Uh oh!

lhotari commented Aug 13, 2025

Uh oh!

lhotari commented Aug 27, 2025

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

8 participants

thetumbled commented Aug 19, 2023 •

edited

Loading

AnonHxy Aug 19, 2023 •

edited

Loading

thetumbled Aug 21, 2023 •

edited

Loading

thetumbled Aug 22, 2023 •

edited

Loading

poorbarcode Aug 23, 2023 •

edited

Loading

thetumbled commented Aug 22, 2023 •

edited

Loading