Skip to content

[Branch 2.7] Fix TopicPoliciesCacheNotInitException issue. #14293

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

Merged
codelipenghui merged 6 commits into from
Feb 16, 2022
Merged

[Branch 2.7] Fix TopicPoliciesCacheNotInitException issue. #14293

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
209e424
[Branch 2.7] Fix get topic policy timeout.
mattisonchao Feb 15, 2022
957d41c
[Branch 2.7] Fix get topic policy timeout.
mattisonchao Feb 15, 2022
d7871e2
Revert "[Branch 2.7] Fix get topic policy timeout."
mattisonchao Feb 15, 2022
2efc3e2
Revert "[Branch 2.7] Fix get topic policy timeout."
mattisonchao Feb 15, 2022
aa26840
[Branch 2.7] Fix get topic policy timeout.
mattisonchao Feb 15, 2022
b44968c
Add test
mattisonchao Feb 15, 2022
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
46 changes: 31 additions & 15 deletions .../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 @@ -144,13 +144,36 @@ public boolean cacheIsInitialized(TopicName topicName) {

@Override
public TopicPolicies getTopicPolicies(TopicName topicName) throws TopicPoliciesCacheNotInitException {
if (!policyCacheInitMap.containsKey(topicName.getNamespaceObject())) {
NamespaceName namespace = topicName.getNamespaceObject();
prepareInitPoliciesCache(namespace, new CompletableFuture<>());
}
if (policyCacheInitMap.containsKey(topicName.getNamespaceObject())
&& !policyCacheInitMap.get(topicName.getNamespaceObject())) {
throw new TopicPoliciesCacheNotInitException();
}
return policiesCache.get(TopicName.get(topicName.getPartitionedTopicName()));
}

private void prepareInitPoliciesCache(NamespaceName namespace, CompletableFuture<Void> result) {
if (policyCacheInitMap.putIfAbsent(namespace, false) == null) {
CompletableFuture<SystemTopicClient.Reader> readerCompletableFuture = namespaceEventsSystemTopicFactory
.createTopicPoliciesSystemTopicClient(namespace).newReaderAsync();
readerCaches.put(namespace, readerCompletableFuture);
readerCompletableFuture.whenComplete((reader, ex) -> {
if (ex != null) {
log.error("[{}] Failed to create reader on __change_events topic", namespace, ex);
result.completeExceptionally(ex);
readerCaches.remove(namespace);
reader.closeAsync();
} else {
initPolicesCache(reader, result);
result.thenRun(() -> readMorePolicies(reader));
}
});
}
}

@Override
public CompletableFuture<TopicPolicies> getTopicPoliciesBypassCacheAsync(TopicName topicName) {
CompletableFuture<TopicPolicies> result = new CompletableFuture<>();
Expand Down Expand Up @@ -180,21 +203,8 @@ public CompletableFuture<Void> addOwnedNamespaceBundleAsync(NamespaceBundle name
ownedBundlesCountPerNamespace.get(namespace).incrementAndGet();
result.complete(null);
} else {
SystemTopicClient systemTopicClient = namespaceEventsSystemTopicFactory.createSystemTopic(namespace
, EventType.TOPIC_POLICY);
ownedBundlesCountPerNamespace.putIfAbsent(namespace, new AtomicInteger(1));
policyCacheInitMap.put(namespace, false);
CompletableFuture<SystemTopicClient.Reader> readerCompletableFuture = systemTopicClient.newReaderAsync();
readerCaches.put(namespace, readerCompletableFuture);
readerCompletableFuture.whenComplete((reader, ex) -> {
if (ex != null) {
log.error("[{}] Failed to create reader on __change_events topic", namespace, ex);
result.completeExceptionally(ex);
} else {
initPolicesCache(reader, result);
result.thenRun(() -> readMorePolicies(reader));
}
});
prepareInitPoliciesCache(namespace, result);
}
}
return result;
Expand Down Expand Up @@ -249,14 +259,20 @@ private void initPolicesCache(SystemTopicClient.Reader reader, CompletableFuture
reader.getSystemTopic().getTopicName(), ex);
future.completeExceptionally(ex);
readerCaches.remove(reader.getSystemTopic().getTopicName().getNamespaceObject());
policyCacheInitMap.remove(reader.getSystemTopic().getTopicName().getNamespaceObject());
reader.closeAsync();
return;
}
if (hasMore) {
reader.readNextAsync().whenComplete((msg, e) -> {
if (e != null) {
log.error("[{}] Failed to read event from the system topic.",
reader.getSystemTopic().getTopicName(), ex);
reader.getSystemTopic().getTopicName(), e);
future.completeExceptionally(e);
readerCaches.remove(reader.getSystemTopic().getTopicName().getNamespaceObject());
policyCacheInitMap.remove(reader.getSystemTopic().getTopicName().getNamespaceObject());
reader.closeAsync();
return;
}
refreshTopicPoliciesCache(msg);
if (log.isDebugEnabled()) {
Expand Down
37 changes: 36 additions & 1 deletion .../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 @@ -22,6 +22,7 @@
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import org.apache.pulsar.broker.service.BrokerServiceException.TopicPoliciesCacheNotInitException;
import org.apache.pulsar.broker.systopic.NamespaceEventsSystemTopicFactory;
import org.apache.pulsar.broker.systopic.SystemTopicClient;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
Expand All @@ -33,8 +34,13 @@
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

import java.lang.reflect.Field;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class SystemTopicBasedTopicPoliciesServiceTest extends MockedPulsarServiceBaseTest {

Expand Down Expand Up @@ -193,4 +199,33 @@ private void prepareData() throws PulsarAdminException {
systemTopicFactory = new NamespaceEventsSystemTopicFactory(pulsarClient);
systemTopicBasedTopicPoliciesService = (SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService();
}

@Test
public void testGetTopicPoliciesWithRetry() throws Exception {
Field initMapField = SystemTopicBasedTopicPoliciesService.class.getDeclaredField("policyCacheInitMap");
initMapField.setAccessible(true);
Map<NamespaceName, Boolean> initMap = (Map)initMapField.get(systemTopicBasedTopicPoliciesService);
initMap.remove(NamespaceName.get(NAMESPACE1));
Field readerCaches = SystemTopicBasedTopicPoliciesService.class.getDeclaredField("readerCaches");
readerCaches.setAccessible(true);
Map<NamespaceName, CompletableFuture<SystemTopicClient.Reader>> readers = (Map)readerCaches.get(systemTopicBasedTopicPoliciesService);
readers.remove(NamespaceName.get(NAMESPACE1));
TopicPolicies initPolicy = TopicPolicies.builder()
.maxConsumerPerTopic(10)
.build();
ScheduledExecutorService executors = Executors.newScheduledThreadPool(1);
executors.schedule(() -> {
try {
systemTopicBasedTopicPoliciesService.updateTopicPoliciesAsync(TOPIC1, initPolicy).get();
} catch (Exception ignore) {}
}, 2000, TimeUnit.MILLISECONDS);
Awaitility.await().untilAsserted(() -> {
try {
TopicPolicies topicPolicies = systemTopicBasedTopicPoliciesService.getTopicPolicies(TOPIC1);
Assert.assertNotNull(topicPolicies);
} catch (Exception ex) {
Assert.fail();
}
});
}
}