Skip to content

[fix] [broker] Ensure policy cache is updated before updateTopicPoliciesAsync returns. #21030

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
thetumbled wants to merge 3 commits into from
Closed

[fix] [broker] Ensure policy cache is updated before updateTopicPoliciesAsync returns. #21030

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
3b8fb93
Ensure that the cache is updated before updateTopicPoliciesAsync retu…
thetumbled Aug 19, 2023
7a91103
remove the result blocking queue if topic is removed.
thetumbled Aug 21, 2023
73393a6
add test code.
thetumbled Aug 21, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 39 additions & 1 deletion .../src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,9 +24,11 @@
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.BlockingDeque;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import javax.annotation.Nonnull;
Expand Down Expand Up @@ -84,6 +86,8 @@ public class SystemTopicBasedTopicPoliciesService implements TopicPoliciesServic
@VisibleForTesting
final Map<TopicName, List<TopicPolicyListener<TopicPolicies>>> listeners = new ConcurrentHashMap<>();

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

@codelipenghui codelipenghui Aug 21, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@thetumbled thetumbled Aug 22, 2023 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@poorbarcode poorbarcode Aug 22, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@thetumbled thetumbled Aug 22, 2023

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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);
                                }
                        });

@poorbarcode poorbarcode Aug 23, 2023 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@codelipenghui codelipenghui Sep 5, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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."


public SystemTopicBasedTopicPoliciesService(PulsarService pulsarService) {
this.pulsarService = pulsarService;
this.clusterName = pulsarService.getConfiguration().getClusterName();
Expand All @@ -97,7 +101,10 @@ public CompletableFuture<Void> deleteTopicPoliciesAsync(TopicName topicName) {

@Override
public CompletableFuture<Void> updateTopicPoliciesAsync(TopicName topicName, TopicPolicies policies) {
return sendTopicPolicyEvent(topicName, ActionType.UPDATE, policies);
CompletableFuture<Void> result = new CompletableFuture<>();
results.computeIfAbsent(topicName, k -> new LinkedBlockingDeque<>()).add(result);
Comment thread
thetumbled marked this conversation as resolved.
return sendTopicPolicyEvent(topicName, ActionType.UPDATE, policies)
.thenCombine(result, (__, ___) -> null);
}

private CompletableFuture<Void> sendTopicPolicyEvent(TopicName topicName, ActionType actionType,
Expand Down Expand Up @@ -420,11 +427,37 @@ private void cleanCacheAndCloseReader(@Nonnull NamespaceName namespace, boolean
});
}

private void completeResults(TopicName topicName) {
if (results.get(topicName) != null) {
BlockingDeque<CompletableFuture> futures = results.get(topicName);
CompletableFuture<Void> future;
while (true) {
future = futures.poll();
if (future == null) {
break;
}
future.complete(null);
}
}
}

private void completeResults(BlockingDeque<CompletableFuture> futures) {
CompletableFuture<Void> future;
while (true) {
future = futures.poll();
if (future == null) {
break;
}
future.complete(null);
}
}

private void readMorePolicies(SystemTopicClient.Reader<PulsarEvent> reader) {
reader.readNextAsync()
.thenAccept(msg -> {
refreshTopicPoliciesCache(msg);
notifyListener(msg);
completeResults(TopicName.get(TopicName.get(msg.getKey()).getPartitionedTopicName()));

@AnonHxy AnonHxy Aug 19, 2023 •
edited
Loading

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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:

  1. setA is waiting for futureA completed. The eventMsg is eventA.
  2. setB is waiting for futureB completed. The eventMsg is eventB.
  3. setC is waiting for futureC completed. The eventMsg is eventC.
  4. The method completeResults will complete futureA and futureB and futureC when read eventA. Howerver setB has not been refresh to the policies cache.
  5. So setC maybe read the old B value

@thetumbled thetumbled Aug 21, 2023 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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:

  1. add a warning to users, that the topic policy setting api can not handle with concurrent cases
  2. or, i can implement a stronger concurrent controll.
    What do you think? @AnonHxy @Demogorgon314 @codelipenghui

@AnonHxy AnonHxy Aug 21, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

OK. I see. Synchronize case is enough I think

})
.whenComplete((__, ex) -> {
if (ex == null) {
Expand Down Expand Up @@ -627,6 +660,11 @@ public void unregisterListener(TopicName topicName, TopicPolicyListener<TopicPol
topicListeners.remove(listener);
if (topicListeners.isEmpty()) {
topicListeners = null;
// clear the policy setting result queue
BlockingDeque<CompletableFuture> futures = results.remove(topicName);
Comment thread
thetumbled marked this conversation as resolved.
if (futures != null) {
completeResults(futures);
}
}
}
return topicListeners;
Expand Down
68 changes: 68 additions & 0 deletions .../test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -239,6 +239,74 @@ public void testGetPolicy() throws ExecutionException, InterruptedException, Top
Assert.assertEquals(policies1, policiesGet1);
}

@Test
public void testUpdateAndGetPolicy() throws ExecutionException, InterruptedException, TopicPoliciesCacheNotInitException {
// Init topic policies
TopicPolicies initPolicy = TopicPolicies.builder()
.maxConsumerPerTopic(10)
.build();
systemTopicBasedTopicPoliciesService.updateTopicPoliciesAsync(TOPIC1, initPolicy).get();

// Wait for all topic policies updated.
Awaitility.await().untilAsserted(() ->
Assert.assertTrue(systemTopicBasedTopicPoliciesService
.getPoliciesCacheInit(TOPIC1.getNamespaceObject())));

// Assert broker is cache all topic policies
Awaitility.await().untilAsserted(() ->
Assert.assertEquals(systemTopicBasedTopicPoliciesService.getTopicPolicies(TOPIC1)
.getMaxConsumerPerTopic().intValue(), 10));

// Update policy for TOPIC1
TopicPolicies policies1 = TopicPolicies.builder()
.maxProducerPerTopic(1)
.build();
systemTopicBasedTopicPoliciesService.updateTopicPoliciesAsync(TOPIC1, policies1).get();

// Ensure that the cache is updated before updateTopicPoliciesAsync returns.
TopicPolicies policiesGet1 = systemTopicBasedTopicPoliciesService.getTopicPolicies(TOPIC1);
Assert.assertEquals(policiesGet1, policies1);
}

@Test
public void testGetPolicyAndRemoveTopic() throws ExecutionException, InterruptedException {
class TopicPolicyListenerImpl implements TopicPolicyListener<TopicPolicies> {

@Override
public void onUpdate(TopicPolicies data) {
//no op.
}
}
TopicPolicyListener<TopicPolicies> listener = new TopicPolicyListenerImpl();
systemTopicBasedTopicPoliciesService.registerListener(TOPIC1, listener);
Assert.assertTrue(systemTopicBasedTopicPoliciesService.listeners.get(TOPIC1).size() >= 1);

// Init topic policies
TopicPolicies initPolicy = TopicPolicies.builder()
.maxConsumerPerTopic(10)
.build();
systemTopicBasedTopicPoliciesService.updateTopicPoliciesAsync(TOPIC1, initPolicy).get();

// Wait for all topic policies updated.
Awaitility.await().untilAsserted(() ->
Assert.assertTrue(systemTopicBasedTopicPoliciesService
.getPoliciesCacheInit(TOPIC1.getNamespaceObject())));

// Assert broker cache all topic policies.
Awaitility.await().untilAsserted(() ->
Assert.assertEquals(systemTopicBasedTopicPoliciesService.getTopicPolicies(TOPIC1)
.getMaxConsumerPerTopic().intValue(), 10));

// Assert policy setting result queue is created.
Assert.assertTrue(systemTopicBasedTopicPoliciesService.results.get(TOPIC1) != null);

// unregister listener and remove the policy setting result queue.
systemTopicBasedTopicPoliciesService.unregisterListener(TOPIC1, listener);
Assert.assertFalse(systemTopicBasedTopicPoliciesService.listeners.containsKey(TOPIC1));
Assert.assertFalse(systemTopicBasedTopicPoliciesService.results.containsKey(TOPIC1));
}


@Test
public void testCacheCleanup() throws Exception {
final String topic = "persistent://" + NAMESPACE1 + "/test" + UUID.randomUUID();
Expand Down