[pulsar-broker] namespace resources use metadata-store api #9351
Conversation
rdhabalia
requested review from
merlimat,
sijie,
jiazhai,
aahmed-se and
codelipenghui
rdhabalia
force-pushed
the
zk_rest_ns
branch
2 times, most recently
from
1afcb3d
to
3c4bb85
Compare
rdhabalia
force-pushed
the
zk_rest_ns
branch
2 times, most recently
from
cd0d738
to
2bc71f9
Compare
|
/pulsarbot run-failure-checks |
| if (maxPartitions > 0 && autoTopicCreationOverride.defaultNumPartitions > maxPartitions) { | ||
| throw new RestException(Status.NOT_ACCEPTABLE, | ||
| "Number of partitions should be less than or equal to " + maxPartitions); | ||
| if (autoTopicCreationOverride != null) { |
There was a problem hiding this comment.
this if statement was not present before, can it be null here ?
There was a problem hiding this comment.
yes, I merged two methods which had duplicate code:
internalRemoveAutoTopicCreation and internalSetAutoSubscriptionCreation
| @@ -2041,4 +2044,9 @@ public void testCompactionStatus() throws Exception { | |||
| assertTrue(admin.topics().compactionStatus(topicName) | |||
| .lastError.contains("Failed at something")); | |||
| } | |||
|
|
|||
| private void clearCache() { | |||
| // (MetadataCacheImpl<ClusterData>) pulsar.getPulsarResources(). | |||
| if (ex.getCause() instanceof BadVersionException) { | ||
| // if resource is updated by other than metadata-cache then metadata-cache will get bad-version | ||
| // exception. so, try to invalidate the cache and try one more time. | ||
| objCache.synchronous().invalidate(key); |
There was a problem hiding this comment.
can we call here this synchronous operation ?
do we risk a deadlock ?
There was a problem hiding this comment.
no, it just cleans up the cache synchronously.
rdhabalia
force-pushed
the
zk_rest_ns
branch
2 times, most recently
from
61001b2
to
38481a9
Compare
| @@ -118,7 +120,7 @@ private MetadataCacheImpl(MetadataStore store, MetadataSerde<T> serde) { | |||
|
|
|||
| @Override | |||
| public CompletableFuture<Void> readModifyUpdateOrCreate(String path, Function<Optional<T>, T> modifyFunction) { | |||
| return objCache.get(path) | |||
| return executeWithRetry(() -> objCache.get(path) | |||
There was a problem hiding this comment.
This part is already in master, can you merge or rebase?
| @@ -418,7 +382,7 @@ protected Policies getNamespacePolicies(NamespaceName namespaceName) { | |||
| try { | |||
| final String namespace = namespaceName.toString(); | |||
| final String policyPath = AdminResource.path(POLICIES, namespace); | |||
| Policies policies = policiesCache().get(policyPath) | |||
| Policies policies = namespaceResources().get(policyPath) | |||
There was a problem hiding this comment.
nit: ideally namespace resources should take a namespace name and internally resolve the path on metadata store.
There was a problem hiding this comment.
yes, make sense. I will try to handle in another PR so, I can handle unit-test failures related to this change separately.
| } catch (KeeperException.NodeExistsException e) { | ||
| log.warn("[{}] Failed to create namespace {} - already exists", clientAppId(), namespaceName); | ||
| } catch (NotFoundException e) { | ||
| log.error("[{}] namespace already exists {}", clientAppId(), namespaceName, e); |
There was a problem hiding this comment.
The exception and the log message seem to have diverged now
| * @param property the property name | ||
| * @return the list of namespaces | ||
| */ | ||
| protected List<String> getListOfNamespaces(String property) throws Exception { |
There was a problem hiding this comment.
| protected List<String> getListOfNamespaces(String property) throws Exception { | |
| protected List<String> getListOfNamespaces(String tenant) throws Exception { |
fix build fix api Fix test: fix test fix test
|
/pulsarbot run-failure-checks |
Motivation
This PR is on top of #9338, with this change, namespace-resources will start using metadata-api and remove zk-dependency.