From 428c18c8d0c3d135189920740192982e11ffb2bf Mon Sep 17 00:00:00 2001 From: Qiang Zhao Date: Wed, 8 Nov 2023 12:22:36 +0800 Subject: [PATCH] [fix][broker] Fix namespace bundle stuck in unloading status (#21445) ### Motivation PR https://github.com/apache/pulsar/pull/21231 made user topic creation rely on system topic `__change_event` if the user is enabling `topicLevelPoliciesEnabled`. It will introduce a race condition with namespace bundle unloading. All creating topics want to create `__change_event` reader but constantly fail by namespace-bundle inactive and retry mechanism. Unfortunately, the namespace-bundle unloading operation is waiting for all the topics to be completed and then release inactive status. Therefore, they will be stuck in a deadlock until one gets a timeout. ### Modifications - Get the topic policy before loading. --- .../pulsar/broker/service/BrokerService.java | 368 ++++++++++-------- .../SystemTopicBasedTopicPoliciesService.java | 19 +- .../broker/admin/TopicAutoCreationTest.java | 8 +- .../namespace/NamespaceUnloadingTest.java | 29 ++ .../broker/service/BrokerServiceTest.java | 4 +- ...temTopicBasedTopicPoliciesServiceTest.java | 29 -- .../client/cli/PulsarClientToolTest.java | 3 + 7 files changed, 244 insertions(+), 216 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java index da556e4422fe3..4f64d5aab869c 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java @@ -49,6 +49,7 @@ import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.concurrent.CancellationException; @@ -70,6 +71,7 @@ import java.util.function.Consumer; import java.util.function.Predicate; import javax.annotation.Nonnull; +import javax.annotation.Nullable; import lombok.AccessLevel; import lombok.AllArgsConstructor; import lombok.Getter; @@ -1052,19 +1054,32 @@ public CompletableFuture> getTopic(final TopicName topicName, bo } final boolean isPersistentTopic = topicName.getDomain().equals(TopicDomain.persistent); if (isPersistentTopic) { - return topics.computeIfAbsent(topicName.toString(), (tpName) -> { - if (topicName.isPartitioned()) { - return fetchPartitionedTopicMetadataAsync(TopicName.get(topicName.getPartitionedTopicName())) - .thenCompose((metadata) -> { - // Allow crate non-partitioned persistent topic that name includes `partition` - if (metadata.partitions == 0 - || topicName.getPartitionIndex() < metadata.partitions) { - return loadOrCreatePersistentTopic(tpName, createIfMissing, properties); - } - return CompletableFuture.completedFuture(Optional.empty()); - }); - } - return loadOrCreatePersistentTopic(tpName, createIfMissing, properties); + final CompletableFuture> topicPoliciesFuture = + getTopicPoliciesBypassSystemTopic(topicName); + return topicPoliciesFuture.exceptionally(ex -> { + final Throwable rc = FutureUtil.unwrapCompletionException(ex); + final String errorInfo = String.format("Topic creation encountered an exception by initialize" + + " topic policies service. topic_name=%s error_message=%s", topicName, rc.getMessage()); + log.error(errorInfo, rc); + throw FutureUtil.wrapToCompletionException(new ServiceUnitNotReadyException(errorInfo)); + }).thenCompose(optionalTopicPolicies -> { + final TopicPolicies topicPolicies = optionalTopicPolicies.orElse(null); + return topics.computeIfAbsent(topicName.toString(), (tpName) -> { + if (topicName.isPartitioned()) { + final TopicName topicNameEntity = TopicName.get(topicName.getPartitionedTopicName()); + return fetchPartitionedTopicMetadataAsync(topicNameEntity) + .thenCompose((metadata) -> { + // Allow crate non-partitioned persistent topic that name includes `partition` + if (metadata.partitions == 0 + || topicName.getPartitionIndex() < metadata.partitions) { + return loadOrCreatePersistentTopic(tpName, createIfMissing, + properties, topicPolicies); + } + return CompletableFuture.completedFuture(Optional.empty()); + }); + } + return loadOrCreatePersistentTopic(tpName, createIfMissing, properties, topicPolicies); + }); }); } else { return topics.computeIfAbsent(topicName.toString(), (name) -> { @@ -1118,6 +1133,18 @@ public CompletableFuture> getTopic(final TopicName topicName, bo } } + private CompletableFuture> getTopicPoliciesBypassSystemTopic(@Nonnull TopicName topicName) { + Objects.requireNonNull(topicName); + final ServiceConfiguration serviceConfiguration = pulsar.getConfiguration(); + if (serviceConfiguration.isSystemTopicEnabled() && serviceConfiguration.isTopicLevelPoliciesEnabled() + && !NamespaceService.isSystemServiceNamespace(topicName.getNamespace()) + && !SystemTopicNames.isTopicPoliciesSystemTopic(topicName.toString())) { + return pulsar.getTopicPoliciesService().getTopicPoliciesAsync(topicName); + } else { + return CompletableFuture.completedFuture(Optional.empty()); + } + } + public CompletableFuture deleteTopic(String topic, boolean forceDelete) { topicEventsDispatcher.notify(topic, TopicEvent.DELETE, EventStage.BEFORE); CompletableFuture result = deleteTopicInternal(topic, forceDelete); @@ -1507,7 +1534,7 @@ public PulsarAdmin getClusterPulsarAdmin(String cluster, Optional c * @throws RuntimeException */ protected CompletableFuture> loadOrCreatePersistentTopic(final String topic, - boolean createIfMissing, Map properties) throws RuntimeException { + boolean createIfMissing, Map properties, @Nullable TopicPolicies topicPolicies) { final CompletableFuture> topicFuture = FutureUtil.createFutureWithTimeout( Duration.ofSeconds(pulsar.getConfiguration().getTopicLoadTimeoutSeconds()), executor(), () -> FAILED_TO_LOAD_TOPIC_TIMEOUT_EXCEPTION); @@ -1525,7 +1552,8 @@ protected CompletableFuture> loadOrCreatePersistentTopic(final S final Semaphore topicLoadSemaphore = topicLoadRequestSemaphore.get(); if (topicLoadSemaphore.tryAcquire()) { - checkOwnershipAndCreatePersistentTopic(topic, createIfMissing, topicFuture, properties); + checkOwnershipAndCreatePersistentTopic(topic, createIfMissing, topicFuture, + properties, topicPolicies); topicFuture.handle((persistentTopic, ex) -> { // release permit and process pending topic topicLoadSemaphore.release(); @@ -1540,7 +1568,7 @@ protected CompletableFuture> loadOrCreatePersistentTopic(final S }); } else { pendingTopicLoadingQueue.add(new TopicLoadingContext(topic, - createIfMissing, topicFuture, properties)); + createIfMissing, topicFuture, properties, topicPolicies)); if (log.isDebugEnabled()) { log.debug("topic-loading for {} added into pending queue", topic); } @@ -1581,7 +1609,7 @@ protected CompletableFuture> fetchTopicPropertiesAsync(Topic private void checkOwnershipAndCreatePersistentTopic(final String topic, boolean createIfMissing, CompletableFuture> topicFuture, - Map properties) { + Map properties, @Nullable TopicPolicies topicPolicies) { TopicName topicName = TopicName.get(topic); pulsar.getNamespaceService().isServiceUnitActiveAsync(topicName) .thenAccept(isActive -> { @@ -1595,7 +1623,8 @@ private void checkOwnershipAndCreatePersistentTopic(final String topic, boolean } propertiesFuture.thenAccept(finalProperties -> //TODO add topicName in properties? - createPersistentTopic(topic, createIfMissing, topicFuture, finalProperties) + createPersistentTopic(topic, createIfMissing, topicFuture, + finalProperties, topicPolicies) ).exceptionally(throwable -> { log.warn("[{}] Read topic property failed", topic, throwable); pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture)); @@ -1620,12 +1649,12 @@ private void checkOwnershipAndCreatePersistentTopic(final String topic, boolean public void createPersistentTopic0(final String topic, boolean createIfMissing, CompletableFuture> topicFuture, Map properties) { - createPersistentTopic(topic, createIfMissing, topicFuture, properties); + createPersistentTopic(topic, createIfMissing, topicFuture, properties, null); } private void createPersistentTopic(final String topic, boolean createIfMissing, CompletableFuture> topicFuture, - Map properties) { + Map properties, @Nullable TopicPolicies topicPolicies) { TopicName topicName = TopicName.get(topic); final long topicCreateTimeMs = TimeUnit.NANOSECONDS.toMillis(System.nanoTime()); @@ -1648,7 +1677,8 @@ private void createPersistentTopic(final String topic, boolean createIfMissing, CompletableFuture isTopicAlreadyMigrated = checkTopicAlreadyMigrated(topicName); - maxTopicsCheck.thenCompose(__ -> isTopicAlreadyMigrated).thenCompose(__ -> getManagedLedgerConfig(topicName)) + maxTopicsCheck.thenCompose(__ -> isTopicAlreadyMigrated) + .thenCompose(__ -> getManagedLedgerConfig(topicName, topicPolicies)) .thenAccept(managedLedgerConfig -> { if (isBrokerEntryMetadataEnabled() || isBrokerPayloadProcessorEnabled()) { // init managedLedger interceptor @@ -1792,170 +1822,167 @@ private CompletableFuture checkTopicAlreadyMigrated(TopicName topicName) { } public CompletableFuture getManagedLedgerConfig(@Nonnull TopicName topicName) { + final CompletableFuture> topicPoliciesFuture = + getTopicPoliciesBypassSystemTopic(topicName); + return topicPoliciesFuture.thenCompose(optionalTopicPolicies -> + getManagedLedgerConfig(topicName, optionalTopicPolicies.orElse(null))); + } + + private CompletableFuture getManagedLedgerConfig(@Nonnull TopicName topicName, + @Nullable TopicPolicies topicPolicies) { requireNonNull(topicName); NamespaceName namespace = topicName.getNamespaceObject(); ServiceConfiguration serviceConfig = pulsar.getConfiguration(); NamespaceResources nsr = pulsar.getPulsarResources().getNamespaceResources(); LocalPoliciesResources lpr = pulsar.getPulsarResources().getLocalPolicies(); - final CompletableFuture> topicPoliciesFuture; - if (pulsar.getConfig().isTopicLevelPoliciesEnabled() - && !NamespaceService.isSystemServiceNamespace(namespace.toString()) - && !SystemTopicNames.isTopicPoliciesSystemTopic(topicName.toString())) { - topicPoliciesFuture = pulsar.getTopicPoliciesService().getTopicPoliciesAsync(topicName); - } else { - topicPoliciesFuture = CompletableFuture.completedFuture(Optional.empty()); - } - return topicPoliciesFuture.thenCompose(topicPoliciesOptional -> { - final CompletableFuture> nsPolicies = nsr.getPoliciesAsync(namespace); - final CompletableFuture> lcPolicies = lpr.getLocalPoliciesAsync(namespace); - return nsPolicies.thenCombine(lcPolicies, (policies, localPolicies) -> { - PersistencePolicies persistencePolicies = null; - RetentionPolicies retentionPolicies = null; - OffloadPoliciesImpl topicLevelOffloadPolicies = null; - if (topicPoliciesOptional.isPresent()) { - final TopicPolicies topicPolicies = topicPoliciesOptional.get(); - persistencePolicies = topicPolicies.getPersistence(); - retentionPolicies = topicPolicies.getRetentionPolicies(); - topicLevelOffloadPolicies = topicPolicies.getOffloadPolicies(); - } - - if (persistencePolicies == null) { - persistencePolicies = policies.map(p -> p.persistence).orElseGet( - () -> new PersistencePolicies(serviceConfig.getManagedLedgerDefaultEnsembleSize(), - serviceConfig.getManagedLedgerDefaultWriteQuorum(), - serviceConfig.getManagedLedgerDefaultAckQuorum(), - serviceConfig.getManagedLedgerDefaultMarkDeleteRateLimit())); - } + final CompletableFuture> nsPolicies = nsr.getPoliciesAsync(namespace); + final CompletableFuture> lcPolicies = lpr.getLocalPoliciesAsync(namespace); + return nsPolicies.thenCombine(lcPolicies, (policies, localPolicies) -> { + PersistencePolicies persistencePolicies = null; + RetentionPolicies retentionPolicies = null; + OffloadPoliciesImpl topicLevelOffloadPolicies = null; + if (topicPolicies != null) { + persistencePolicies = topicPolicies.getPersistence(); + retentionPolicies = topicPolicies.getRetentionPolicies(); + topicLevelOffloadPolicies = topicPolicies.getOffloadPolicies(); + } - if (retentionPolicies == null) { - retentionPolicies = policies.map(p -> p.retention_policies).orElseGet( - () -> new RetentionPolicies(serviceConfig.getDefaultRetentionTimeInMinutes(), - serviceConfig.getDefaultRetentionSizeInMB()) - ); - } + if (persistencePolicies == null) { + persistencePolicies = policies.map(p -> p.persistence).orElseGet( + () -> new PersistencePolicies(serviceConfig.getManagedLedgerDefaultEnsembleSize(), + serviceConfig.getManagedLedgerDefaultWriteQuorum(), + serviceConfig.getManagedLedgerDefaultAckQuorum(), + serviceConfig.getManagedLedgerDefaultMarkDeleteRateLimit())); + } - ManagedLedgerConfig managedLedgerConfig = new ManagedLedgerConfig(); - managedLedgerConfig.setEnsembleSize(persistencePolicies.getBookkeeperEnsemble()); - managedLedgerConfig.setWriteQuorumSize(persistencePolicies.getBookkeeperWriteQuorum()); - managedLedgerConfig.setAckQuorumSize(persistencePolicies.getBookkeeperAckQuorum()); + if (retentionPolicies == null) { + retentionPolicies = policies.map(p -> p.retention_policies).orElseGet( + () -> new RetentionPolicies(serviceConfig.getDefaultRetentionTimeInMinutes(), + serviceConfig.getDefaultRetentionSizeInMB()) + ); + } - if (serviceConfig.isStrictBookieAffinityEnabled()) { + ManagedLedgerConfig managedLedgerConfig = new ManagedLedgerConfig(); + managedLedgerConfig.setEnsembleSize(persistencePolicies.getBookkeeperEnsemble()); + managedLedgerConfig.setWriteQuorumSize(persistencePolicies.getBookkeeperWriteQuorum()); + managedLedgerConfig.setAckQuorumSize(persistencePolicies.getBookkeeperAckQuorum()); + + if (serviceConfig.isStrictBookieAffinityEnabled()) { + managedLedgerConfig.setBookKeeperEnsemblePlacementPolicyClassName( + IsolatedBookieEnsemblePlacementPolicy.class); + if (localPolicies.isPresent() && localPolicies.get().bookieAffinityGroup != null) { + Map properties = new HashMap<>(); + properties.put(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, + localPolicies.get().bookieAffinityGroup.getBookkeeperAffinityGroupPrimary()); + properties.put(IsolatedBookieEnsemblePlacementPolicy.SECONDARY_ISOLATION_BOOKIE_GROUPS, + localPolicies.get().bookieAffinityGroup.getBookkeeperAffinityGroupSecondary()); + managedLedgerConfig.setBookKeeperEnsemblePlacementPolicyProperties(properties); + } else if (isSystemTopic(topicName)) { + Map properties = new HashMap<>(); + properties.put(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, "*"); + properties.put(IsolatedBookieEnsemblePlacementPolicy + .SECONDARY_ISOLATION_BOOKIE_GROUPS, "*"); + managedLedgerConfig.setBookKeeperEnsemblePlacementPolicyProperties(properties); + } else { + Map properties = new HashMap<>(); + properties.put(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, ""); + properties.put(IsolatedBookieEnsemblePlacementPolicy.SECONDARY_ISOLATION_BOOKIE_GROUPS, ""); + managedLedgerConfig.setBookKeeperEnsemblePlacementPolicyProperties(properties); + } + } else { + if (localPolicies.isPresent() && localPolicies.get().bookieAffinityGroup != null) { managedLedgerConfig.setBookKeeperEnsemblePlacementPolicyClassName( IsolatedBookieEnsemblePlacementPolicy.class); - if (localPolicies.isPresent() && localPolicies.get().bookieAffinityGroup != null) { - Map properties = new HashMap<>(); - properties.put(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, - localPolicies.get().bookieAffinityGroup.getBookkeeperAffinityGroupPrimary()); - properties.put(IsolatedBookieEnsemblePlacementPolicy.SECONDARY_ISOLATION_BOOKIE_GROUPS, - localPolicies.get().bookieAffinityGroup.getBookkeeperAffinityGroupSecondary()); - managedLedgerConfig.setBookKeeperEnsemblePlacementPolicyProperties(properties); - } else if (isSystemTopic(topicName)) { - Map properties = new HashMap<>(); - properties.put(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, "*"); - properties.put(IsolatedBookieEnsemblePlacementPolicy - .SECONDARY_ISOLATION_BOOKIE_GROUPS, "*"); - managedLedgerConfig.setBookKeeperEnsemblePlacementPolicyProperties(properties); - } else { - Map properties = new HashMap<>(); - properties.put(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, ""); - properties.put(IsolatedBookieEnsemblePlacementPolicy.SECONDARY_ISOLATION_BOOKIE_GROUPS, ""); - managedLedgerConfig.setBookKeeperEnsemblePlacementPolicyProperties(properties); - } - } else { - if (localPolicies.isPresent() && localPolicies.get().bookieAffinityGroup != null) { - managedLedgerConfig.setBookKeeperEnsemblePlacementPolicyClassName( - IsolatedBookieEnsemblePlacementPolicy.class); - Map properties = new HashMap<>(); - properties.put(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, - localPolicies.get().bookieAffinityGroup.getBookkeeperAffinityGroupPrimary()); - properties.put(IsolatedBookieEnsemblePlacementPolicy.SECONDARY_ISOLATION_BOOKIE_GROUPS, - localPolicies.get().bookieAffinityGroup.getBookkeeperAffinityGroupSecondary()); - managedLedgerConfig.setBookKeeperEnsemblePlacementPolicyProperties(properties); - } + Map properties = new HashMap<>(); + properties.put(IsolatedBookieEnsemblePlacementPolicy.ISOLATION_BOOKIE_GROUPS, + localPolicies.get().bookieAffinityGroup.getBookkeeperAffinityGroupPrimary()); + properties.put(IsolatedBookieEnsemblePlacementPolicy.SECONDARY_ISOLATION_BOOKIE_GROUPS, + localPolicies.get().bookieAffinityGroup.getBookkeeperAffinityGroupSecondary()); + managedLedgerConfig.setBookKeeperEnsemblePlacementPolicyProperties(properties); } + } - managedLedgerConfig.setThrottleMarkDelete(persistencePolicies.getManagedLedgerMaxMarkDeleteRate()); - managedLedgerConfig.setDigestType(serviceConfig.getManagedLedgerDigestType()); - managedLedgerConfig.setPassword(serviceConfig.getManagedLedgerPassword()); - - managedLedgerConfig - .setMaxUnackedRangesToPersist(serviceConfig.getManagedLedgerMaxUnackedRangesToPersist()); - managedLedgerConfig.setPersistentUnackedRangesWithMultipleEntriesEnabled( - serviceConfig.isPersistentUnackedRangesWithMultipleEntriesEnabled()); - managedLedgerConfig.setMaxUnackedRangesToPersistInMetadataStore( - serviceConfig.getManagedLedgerMaxUnackedRangesToPersistInMetadataStore()); - managedLedgerConfig.setMaxEntriesPerLedger(serviceConfig.getManagedLedgerMaxEntriesPerLedger()); - managedLedgerConfig - .setMinimumRolloverTime(serviceConfig.getManagedLedgerMinLedgerRolloverTimeMinutes(), - TimeUnit.MINUTES); - managedLedgerConfig - .setMaximumRolloverTime(serviceConfig.getManagedLedgerMaxLedgerRolloverTimeMinutes(), - TimeUnit.MINUTES); - managedLedgerConfig.setMaxSizePerLedgerMb(serviceConfig.getManagedLedgerMaxSizePerLedgerMbytes()); - - managedLedgerConfig.setMetadataOperationsTimeoutSeconds( - serviceConfig.getManagedLedgerMetadataOperationsTimeoutSeconds()); - managedLedgerConfig - .setReadEntryTimeoutSeconds(serviceConfig.getManagedLedgerReadEntryTimeoutSeconds()); - managedLedgerConfig - .setAddEntryTimeoutSeconds(serviceConfig.getManagedLedgerAddEntryTimeoutSeconds()); - managedLedgerConfig.setMetadataEnsembleSize(serviceConfig.getManagedLedgerDefaultEnsembleSize()); - managedLedgerConfig.setUnackedRangesOpenCacheSetEnabled( - serviceConfig.isManagedLedgerUnackedRangesOpenCacheSetEnabled()); - managedLedgerConfig.setMetadataWriteQuorumSize(serviceConfig.getManagedLedgerDefaultWriteQuorum()); - managedLedgerConfig.setMetadataAckQuorumSize(serviceConfig.getManagedLedgerDefaultAckQuorum()); - managedLedgerConfig - .setMetadataMaxEntriesPerLedger(serviceConfig.getManagedLedgerCursorMaxEntriesPerLedger()); - - managedLedgerConfig - .setLedgerRolloverTimeout(serviceConfig.getManagedLedgerCursorRolloverTimeInSeconds()); - managedLedgerConfig - .setRetentionTime(retentionPolicies.getRetentionTimeInMinutes(), TimeUnit.MINUTES); - managedLedgerConfig.setRetentionSizeInMB(retentionPolicies.getRetentionSizeInMB()); - managedLedgerConfig.setAutoSkipNonRecoverableData(serviceConfig.isAutoSkipNonRecoverableData()); - managedLedgerConfig.setLazyCursorRecovery(serviceConfig.isLazyCursorRecovery()); - managedLedgerConfig.setInactiveLedgerRollOverTime( - serviceConfig.getManagedLedgerInactiveLedgerRolloverTimeSeconds(), TimeUnit.SECONDS); - managedLedgerConfig.setCacheEvictionByMarkDeletedPosition( - serviceConfig.isCacheEvictionByMarkDeletedPosition()); - managedLedgerConfig.setMinimumBacklogCursorsForCaching( - serviceConfig.getManagedLedgerMinimumBacklogCursorsForCaching()); - managedLedgerConfig.setMinimumBacklogEntriesForCaching( - serviceConfig.getManagedLedgerMinimumBacklogEntriesForCaching()); - managedLedgerConfig.setMaxBacklogBetweenCursorsForCaching( - serviceConfig.getManagedLedgerMaxBacklogBetweenCursorsForCaching()); - - OffloadPoliciesImpl nsLevelOffloadPolicies = - (OffloadPoliciesImpl) policies.map(p -> p.offload_policies).orElse(null); - OffloadPoliciesImpl offloadPolicies = OffloadPoliciesImpl.mergeConfiguration( - topicLevelOffloadPolicies, - OffloadPoliciesImpl.oldPoliciesCompatible(nsLevelOffloadPolicies, policies.orElse(null)), - getPulsar().getConfig().getProperties()); - if (NamespaceService.isSystemServiceNamespace(namespace.toString())) { - managedLedgerConfig.setLedgerOffloader(NullLedgerOffloader.INSTANCE); - } else { - if (topicLevelOffloadPolicies != null) { - try { - LedgerOffloader topicLevelLedgerOffLoader = - pulsar().createManagedLedgerOffloader(offloadPolicies); - managedLedgerConfig.setLedgerOffloader(topicLevelLedgerOffLoader); - } catch (PulsarServerException e) { - throw new RuntimeException(e); - } - } else { - //If the topic level policy is null, use the namespace level - managedLedgerConfig - .setLedgerOffloader(pulsar.getManagedLedgerOffloader(namespace, offloadPolicies)); + managedLedgerConfig.setThrottleMarkDelete(persistencePolicies.getManagedLedgerMaxMarkDeleteRate()); + managedLedgerConfig.setDigestType(serviceConfig.getManagedLedgerDigestType()); + managedLedgerConfig.setPassword(serviceConfig.getManagedLedgerPassword()); + + managedLedgerConfig + .setMaxUnackedRangesToPersist(serviceConfig.getManagedLedgerMaxUnackedRangesToPersist()); + managedLedgerConfig.setPersistentUnackedRangesWithMultipleEntriesEnabled( + serviceConfig.isPersistentUnackedRangesWithMultipleEntriesEnabled()); + managedLedgerConfig.setMaxUnackedRangesToPersistInMetadataStore( + serviceConfig.getManagedLedgerMaxUnackedRangesToPersistInMetadataStore()); + managedLedgerConfig.setMaxEntriesPerLedger(serviceConfig.getManagedLedgerMaxEntriesPerLedger()); + managedLedgerConfig + .setMinimumRolloverTime(serviceConfig.getManagedLedgerMinLedgerRolloverTimeMinutes(), + TimeUnit.MINUTES); + managedLedgerConfig + .setMaximumRolloverTime(serviceConfig.getManagedLedgerMaxLedgerRolloverTimeMinutes(), + TimeUnit.MINUTES); + managedLedgerConfig.setMaxSizePerLedgerMb(serviceConfig.getManagedLedgerMaxSizePerLedgerMbytes()); + + managedLedgerConfig.setMetadataOperationsTimeoutSeconds( + serviceConfig.getManagedLedgerMetadataOperationsTimeoutSeconds()); + managedLedgerConfig + .setReadEntryTimeoutSeconds(serviceConfig.getManagedLedgerReadEntryTimeoutSeconds()); + managedLedgerConfig + .setAddEntryTimeoutSeconds(serviceConfig.getManagedLedgerAddEntryTimeoutSeconds()); + managedLedgerConfig.setMetadataEnsembleSize(serviceConfig.getManagedLedgerDefaultEnsembleSize()); + managedLedgerConfig.setUnackedRangesOpenCacheSetEnabled( + serviceConfig.isManagedLedgerUnackedRangesOpenCacheSetEnabled()); + managedLedgerConfig.setMetadataWriteQuorumSize(serviceConfig.getManagedLedgerDefaultWriteQuorum()); + managedLedgerConfig.setMetadataAckQuorumSize(serviceConfig.getManagedLedgerDefaultAckQuorum()); + managedLedgerConfig + .setMetadataMaxEntriesPerLedger(serviceConfig.getManagedLedgerCursorMaxEntriesPerLedger()); + + managedLedgerConfig + .setLedgerRolloverTimeout(serviceConfig.getManagedLedgerCursorRolloverTimeInSeconds()); + managedLedgerConfig + .setRetentionTime(retentionPolicies.getRetentionTimeInMinutes(), TimeUnit.MINUTES); + managedLedgerConfig.setRetentionSizeInMB(retentionPolicies.getRetentionSizeInMB()); + managedLedgerConfig.setAutoSkipNonRecoverableData(serviceConfig.isAutoSkipNonRecoverableData()); + managedLedgerConfig.setLazyCursorRecovery(serviceConfig.isLazyCursorRecovery()); + managedLedgerConfig.setInactiveLedgerRollOverTime( + serviceConfig.getManagedLedgerInactiveLedgerRolloverTimeSeconds(), TimeUnit.SECONDS); + managedLedgerConfig.setCacheEvictionByMarkDeletedPosition( + serviceConfig.isCacheEvictionByMarkDeletedPosition()); + managedLedgerConfig.setMinimumBacklogCursorsForCaching( + serviceConfig.getManagedLedgerMinimumBacklogCursorsForCaching()); + managedLedgerConfig.setMinimumBacklogEntriesForCaching( + serviceConfig.getManagedLedgerMinimumBacklogEntriesForCaching()); + managedLedgerConfig.setMaxBacklogBetweenCursorsForCaching( + serviceConfig.getManagedLedgerMaxBacklogBetweenCursorsForCaching()); + + OffloadPoliciesImpl nsLevelOffloadPolicies = + (OffloadPoliciesImpl) policies.map(p -> p.offload_policies).orElse(null); + OffloadPoliciesImpl offloadPolicies = OffloadPoliciesImpl.mergeConfiguration( + topicLevelOffloadPolicies, + OffloadPoliciesImpl.oldPoliciesCompatible(nsLevelOffloadPolicies, policies.orElse(null)), + getPulsar().getConfig().getProperties()); + if (NamespaceService.isSystemServiceNamespace(namespace.toString())) { + managedLedgerConfig.setLedgerOffloader(NullLedgerOffloader.INSTANCE); + } else { + if (topicLevelOffloadPolicies != null) { + try { + LedgerOffloader topicLevelLedgerOffLoader = + pulsar().createManagedLedgerOffloader(offloadPolicies); + managedLedgerConfig.setLedgerOffloader(topicLevelLedgerOffLoader); + } catch (PulsarServerException e) { + throw new RuntimeException(e); } + } else { + //If the topic level policy is null, use the namespace level + managedLedgerConfig + .setLedgerOffloader(pulsar.getManagedLedgerOffloader(namespace, offloadPolicies)); } + } - managedLedgerConfig.setDeletionAtBatchIndexLevelEnabled( - serviceConfig.isAcknowledgmentAtBatchIndexLevelEnabled()); - managedLedgerConfig.setNewEntriesCheckDelayInMillis( - serviceConfig.getManagedLedgerNewEntriesCheckDelayInMillis()); - return managedLedgerConfig; - }); + managedLedgerConfig.setDeletionAtBatchIndexLevelEnabled( + serviceConfig.isAcknowledgmentAtBatchIndexLevelEnabled()); + managedLedgerConfig.setNewEntriesCheckDelayInMillis( + serviceConfig.getManagedLedgerNewEntriesCheckDelayInMillis()); + return managedLedgerConfig; }); } @@ -3083,7 +3110,7 @@ private void createPendingLoadTopic() { checkOwnershipAndCreatePersistentTopic(topic, pendingTopic.isCreateIfMissing(), pendingFuture, - pendingTopic.getProperties()); + pendingTopic.getProperties(), pendingTopic.getTopicPolicies()); pendingFuture.handle((persistentTopic, ex) -> { // release permit and process next pending topic if (acquiredPermit) { @@ -3655,5 +3682,6 @@ private static class TopicLoadingContext { private final boolean createIfMissing; private final CompletableFuture> topicFuture; private final Map properties; + private final TopicPolicies topicPolicies; } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java index da31234095446..80fecbe67b646 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesService.java @@ -29,7 +29,6 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; -import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import javax.annotation.Nonnull; import org.apache.commons.lang3.tuple.MutablePair; @@ -43,10 +42,8 @@ import org.apache.pulsar.client.api.Message; import org.apache.pulsar.client.api.MessageId; import org.apache.pulsar.client.api.PulsarClientException; -import org.apache.pulsar.client.impl.Backoff; import org.apache.pulsar.client.impl.MessageImpl; import org.apache.pulsar.client.impl.TopicMessageImpl; -import org.apache.pulsar.client.util.RetryUtil; import org.apache.pulsar.common.events.ActionType; import org.apache.pulsar.common.events.EventType; import org.apache.pulsar.common.events.PulsarEvent; @@ -320,7 +317,7 @@ public CompletableFuture addOwnedNamespaceBundleAsync(NamespaceBundle name requireNonNull(namespace); return policyCacheInitMap.computeIfAbsent(namespace, (k) -> { final CompletableFuture> readerCompletableFuture = - createSystemTopicClientWithRetry(namespace); + createSystemTopicClient(namespace); readerCaches.put(namespace, readerCompletableFuture); ownedBundlesCountPerNamespace.putIfAbsent(namespace, new AtomicInteger(1)); final CompletableFuture initFuture = readerCompletableFuture @@ -346,20 +343,16 @@ public CompletableFuture addOwnedNamespaceBundleAsync(NamespaceBundle name }); } - protected CompletableFuture> createSystemTopicClientWithRetry( + protected CompletableFuture> createSystemTopicClient( NamespaceName namespace) { - CompletableFuture> result = new CompletableFuture<>(); try { createSystemTopicFactoryIfNeeded(); - } catch (PulsarServerException e) { - result.completeExceptionally(e); - return result; + } catch (PulsarServerException ex) { + return FutureUtil.failedFuture(ex); } - SystemTopicClient systemTopicClient = namespaceEventsSystemTopicFactory + final SystemTopicClient systemTopicClient = namespaceEventsSystemTopicFactory .createTopicPoliciesSystemTopicClient(namespace); - Backoff backoff = new Backoff(1, TimeUnit.SECONDS, 3, TimeUnit.SECONDS, 10, TimeUnit.SECONDS); - RetryUtil.retryAsynchronously(systemTopicClient::newReaderAsync, backoff, pulsarService.getExecutor(), result); - return result; + return systemTopicClient.newReaderAsync(); } @Override diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAutoCreationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAutoCreationTest.java index d447787d1b7cb..590edc2d3f3bb 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAutoCreationTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicAutoCreationTest.java @@ -135,6 +135,8 @@ public void testPartitionedTopicAutoCreationForbiddenDuringNamespaceDeletion() new InetSocketAddress(pulsar.getAdvertisedAddress(), pulsar.getBrokerListenPort().get()); return CompletableFuture.completedFuture(Pair.of(brokerAddress, brokerAddress)); }); + final String topicPoliciesServiceInitException + = "Topic creation encountered an exception by initialize topic policies service"; // Creating a producer and creating a Consumer may trigger automatic topic // creation, let's try to create a Producer and a Consumer @@ -145,7 +147,8 @@ public void testPartitionedTopicAutoCreationForbiddenDuringNamespaceDeletion() } catch (PulsarClientException.LookupException expected) { String msg = "Namespace bundle for topic (%s) not served by this instance"; log.info("Expected error", expected); - assertTrue(expected.getMessage().contains(String.format(msg, topic))); + assertTrue(expected.getMessage().contains(String.format(msg, topic)) + || expected.getMessage().contains(topicPoliciesServiceInitException)); } try (Consumer ignored = pulsarClient.newConsumer() @@ -155,7 +158,8 @@ public void testPartitionedTopicAutoCreationForbiddenDuringNamespaceDeletion() } catch (PulsarClientException.LookupException expected) { String msg = "Namespace bundle for topic (%s) not served by this instance"; log.info("Expected error", expected); - assertTrue(expected.getMessage().contains(String.format(msg, topic))); + assertTrue(expected.getMessage().contains(String.format(msg, topic)) + || expected.getMessage().contains(topicPoliciesServiceInitException)); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceUnloadingTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceUnloadingTest.java index 0dbfe1760879a..1526611874a62 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceUnloadingTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/namespace/NamespaceUnloadingTest.java @@ -22,8 +22,12 @@ import com.google.common.collect.Sets; +import lombok.Cleanup; import org.apache.pulsar.broker.service.BrokerTestBase; +import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.client.api.Producer; +import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.client.api.Schema; import org.testng.annotations.AfterMethod; import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; @@ -34,6 +38,9 @@ public class NamespaceUnloadingTest extends BrokerTestBase { @BeforeMethod @Override protected void setup() throws Exception { + conf.setTopicLevelPoliciesEnabled(true); + conf.setForceDeleteNamespaceAllowed(true); + conf.setTopicLoadTimeoutSeconds(Integer.MAX_VALUE); super.baseSetup(); } @@ -68,4 +75,26 @@ public void testUnloadPartiallyLoadedNamespace() throws Exception { producer.close(); } + @Test + public void testUnloadWithTopicCreation() throws PulsarAdminException, PulsarClientException { + final String namespaceName = "prop/ns_unloading"; + final String topicName = "persistent://prop/ns_unloading/with_topic_creation"; + final int partitions = 5; + admin.namespaces().createNamespace(namespaceName, 1); + admin.topics().createPartitionedTopic(topicName, partitions); + @Cleanup + Producer producer = pulsarClient.newProducer(Schema.INT32) + .topic(topicName) + .create(); + + for (int i = 0; i < 100; i++) { + admin.namespaces().unloadNamespaceBundle(namespaceName, "0x00000000_0xffffffff"); + } + + for (int i = 0; i < partitions; i++) { + producer.send(i); + } + admin.namespaces().deleteNamespace(namespaceName, true); + } + } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java index 520719bd19963..35f68a7a9ca7f 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceTest.java @@ -1160,7 +1160,7 @@ public void testTopicLoadingOnDisableNamespaceBundle() throws Exception { // try to create topic which should fail as bundle is disable CompletableFuture> futureResult = pulsar.getBrokerService() - .loadOrCreatePersistentTopic(topicName, true, null); + .loadOrCreatePersistentTopic(topicName, true, null, null); try { futureResult.get(); @@ -1204,7 +1204,7 @@ public void testConcurrentLoadTopicExceedLimitShouldNotBeAutoCreated() throws Ex for (int i = 0; i < 10; i++) { // try to create topic which should fail as bundle is disable CompletableFuture> futureResult = pulsar.getBrokerService() - .loadOrCreatePersistentTopic(topicName + "_" + i, false, null); + .loadOrCreatePersistentTopic(topicName + "_" + i, false, null, null); loadFutures.add(futureResult); } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java index f0255a13cbd9b..919e82afa6415 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/SystemTopicBasedTopicPoliciesServiceTest.java @@ -18,9 +18,6 @@ */ package org.apache.pulsar.broker.service; -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.Mockito.doReturn; -import static org.mockito.Mockito.mock; import static org.mockito.Mockito.spy; import static org.testng.AssertJUnit.assertEquals; import static org.testng.AssertJUnit.assertNotNull; @@ -44,11 +41,8 @@ import org.apache.commons.lang3.reflect.FieldUtils; 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.broker.systopic.TopicPoliciesSystemTopicClient; import org.apache.pulsar.client.admin.PulsarAdminException; -import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.impl.Backoff; import org.apache.pulsar.client.impl.BackoffBuilder; import org.apache.pulsar.common.events.PulsarEvent; @@ -57,7 +51,6 @@ import org.apache.pulsar.common.policies.data.ClusterData; import org.apache.pulsar.common.policies.data.TenantInfoImpl; import org.apache.pulsar.common.policies.data.TopicPolicies; -import org.apache.pulsar.common.util.FutureUtil; import org.assertj.core.api.Assertions; import org.awaitility.Awaitility; import org.mockito.Mockito; @@ -322,28 +315,6 @@ public void testGetPolicyTimeout() throws Exception { assertTrue("actual:" + cost, cost >= 5000 - 1000); } - @Test - public void testCreatSystemTopicClientWithRetry() throws Exception { - SystemTopicBasedTopicPoliciesService service = - spy((SystemTopicBasedTopicPoliciesService) pulsar.getTopicPoliciesService()); - Field field = SystemTopicBasedTopicPoliciesService.class - .getDeclaredField("namespaceEventsSystemTopicFactory"); - field.setAccessible(true); - NamespaceEventsSystemTopicFactory factory = spy((NamespaceEventsSystemTopicFactory) field.get(service)); - SystemTopicClient client = mock(TopicPoliciesSystemTopicClient.class); - doReturn(client).when(factory).createTopicPoliciesSystemTopicClient(any()); - field.set(service, factory); - - SystemTopicClient.Reader reader = mock(SystemTopicClient.Reader.class); - // Throw an exception first, create successfully after retrying - doReturn(FutureUtil.failedFuture(new PulsarClientException("test"))) - .doReturn(CompletableFuture.completedFuture(reader)).when(client).newReaderAsync(); - - SystemTopicClient.Reader reader1 = service.createSystemTopicClientWithRetry(null).get(); - - assertEquals(reader1, reader); - } - @Test public void testGetTopicPoliciesWithRetry() throws Exception { Field initMapField = SystemTopicBasedTopicPoliciesService.class.getDeclaredField("policyCacheInitMap"); diff --git a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/client/cli/PulsarClientToolTest.java b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/client/cli/PulsarClientToolTest.java index f84f344ba5d69..84b8060b3115d 100644 --- a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/client/cli/PulsarClientToolTest.java +++ b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/client/cli/PulsarClientToolTest.java @@ -123,6 +123,7 @@ public void testNonDurableSubscribe() throws Exception { properties.setProperty("useTls", "false"); final String topicName = getTopicWithRandomSuffix("non-durable"); + admin.topics().createNonPartitionedTopic(topicName); int numberOfMessages = 10; @Cleanup("shutdownNow") @@ -214,6 +215,7 @@ public void testRead() throws Exception { properties.setProperty("useTls", "false"); final String topicName = getTopicWithRandomSuffix("reader"); + admin.topics().createNonPartitionedTopic(topicName); int numberOfMessages = 10; @Cleanup("shutdownNow") @@ -263,6 +265,7 @@ public void testEncryption() throws Exception { properties.setProperty("useTls", "false"); final String topicName = getTopicWithRandomSuffix("encryption"); + admin.topics().createNonPartitionedTopic(topicName); final String keyUriBase = "file:../pulsar-broker/src/test/resources/certificate/"; final int numberOfMessages = 10;