From 177d67f4c8ad2eba72423b788aea429f404cac2e Mon Sep 17 00:00:00 2001 From: Mattison Chao Date: Mon, 13 Nov 2023 10:17:33 +0800 Subject: [PATCH] [fix][broker] Fix namespace bundle stuck in unloading status (#21445) --- .../pulsar/broker/service/BrokerService.java | 330 +++++++++++------- .../SystemTopicBasedTopicPoliciesService.java | 20 +- .../broker/admin/TopicAutoCreationTest.java | 8 +- .../namespace/NamespaceUnloadingTest.java | 29 ++ .../broker/service/BrokerServiceTest.java | 4 +- ...temTopicBasedTopicPoliciesServiceTest.java | 29 -- .../client/cli/PulsarClientToolTest.java | 52 +++ 7 files changed, 303 insertions(+), 169 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 b028bfa3e4f24..e359b41288ac0 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 @@ -53,6 +53,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; @@ -75,6 +76,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; @@ -982,8 +984,32 @@ public CompletableFuture> getTopic(final String topic, boolean c } final boolean isPersistentTopic = TopicName.get(topic).getDomain().equals(TopicDomain.persistent); if (isPersistentTopic) { - return topics.computeIfAbsent(topic, (topicName) -> { - return this.loadOrCreatePersistentTopic(topicName, 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(topic, (name) -> { @@ -1027,6 +1053,18 @@ public CompletableFuture deleteSchemaStorage(String topic) { } } + 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) { return deleteTopic(topic, forceDelete, false); } @@ -1387,7 +1425,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); @@ -1405,7 +1443,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(); @@ -1414,7 +1453,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); } @@ -1429,12 +1468,28 @@ protected CompletableFuture> loadOrCreatePersistentTopic(final S 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 -> { if (isActive) { - createPersistentTopic(topic, createIfMissing, topicFuture, properties); + CompletableFuture> propertiesFuture; + if (properties == null) { + //Read properties from storage when loading topic. + propertiesFuture = fetchTopicPropertiesAsync(topicName); + } else { + propertiesFuture = CompletableFuture.completedFuture(properties); + } + propertiesFuture.thenAccept(finalProperties -> + //TODO add topicName in properties? + createPersistentTopic(topic, createIfMissing, topicFuture, + finalProperties, topicPolicies) + ).exceptionally(throwable -> { + log.warn("[{}] Read topic property failed", topic, throwable); + pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture)); + topicFuture.completeExceptionally(throwable); + return null; + }); } else { // namespace is being unloaded String msg = String.format("Namespace is being unloaded, cannot add topic %s", topic); @@ -1452,12 +1507,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()); @@ -1579,139 +1634,167 @@ public void openLedgerFailed(ManagedLedgerException exception, Object ctx) { } 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()) - && !EventsTopicNames.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(); - } + 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 (persistencePolicies == null) { - persistencePolicies = policies.map(p -> p.persistence).orElseGet( - () -> new PersistencePolicies(serviceConfig.getManagedLedgerDefaultEnsembleSize(), - serviceConfig.getManagedLedgerDefaultWriteQuorum(), - serviceConfig.getManagedLedgerDefaultAckQuorum(), - serviceConfig.getManagedLedgerDefaultMarkDeleteRateLimit())); - } + if (persistencePolicies == null) { + persistencePolicies = policies.map(p -> p.persistence).orElseGet( + () -> new PersistencePolicies(serviceConfig.getManagedLedgerDefaultEnsembleSize(), + serviceConfig.getManagedLedgerDefaultWriteQuorum(), + serviceConfig.getManagedLedgerDefaultAckQuorum(), + serviceConfig.getManagedLedgerDefaultMarkDeleteRateLimit())); + } - if (retentionPolicies == null) { - retentionPolicies = policies.map(p -> p.retention_policies).orElseGet( - () -> new RetentionPolicies(serviceConfig.getDefaultRetentionTimeInMinutes(), - serviceConfig.getDefaultRetentionSizeInMB()) - ); - } + if (retentionPolicies == null) { + retentionPolicies = policies.map(p -> p.retention_policies).orElseGet( + () -> new RetentionPolicies(serviceConfig.getDefaultRetentionTimeInMinutes(), + serviceConfig.getDefaultRetentionSizeInMB()) + ); + } + ManagedLedgerConfig managedLedgerConfig = new ManagedLedgerConfig(); + managedLedgerConfig.setEnsembleSize(persistencePolicies.getBookkeeperEnsemble()); + managedLedgerConfig.setWriteQuorumSize(persistencePolicies.getBookkeeperWriteQuorum()); + managedLedgerConfig.setAckQuorumSize(persistencePolicies.getBookkeeperAckQuorum()); - 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) { - managedLedgerConfig - .setBookKeeperEnsemblePlacementPolicyClassName( - IsolatedBookieEnsemblePlacementPolicy.class); - Map properties = Maps.newHashMap(); + 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); } - managedLedgerConfig.setThrottleMarkDelete(persistencePolicies.getManagedLedgerMaxMarkDeleteRate()); - managedLedgerConfig.setDigestType(serviceConfig.getManagedLedgerDigestType()); - managedLedgerConfig.setPassword(serviceConfig.getManagedLedgerPassword()); - - managedLedgerConfig - .setMaxUnackedRangesToPersist(serviceConfig.getManagedLedgerMaxUnackedRangesToPersist()); - managedLedgerConfig.setMaxUnackedRangesToPersistInZk( - serviceConfig.getManagedLedgerMaxUnackedRangesToPersistInZooKeeper()); - 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()); - - 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)); - } + } 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); } + } - managedLedgerConfig.setDeletionAtBatchIndexLevelEnabled( - serviceConfig.isAcknowledgmentAtBatchIndexLevelEnabled()); - managedLedgerConfig.setNewEntriesCheckDelayInMillis( - serviceConfig.getManagedLedgerNewEntriesCheckDelayInMillis()); + 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)); + } + } - return managedLedgerConfig; - }); + managedLedgerConfig.setDeletionAtBatchIndexLevelEnabled( + serviceConfig.isAcknowledgmentAtBatchIndexLevelEnabled()); + managedLedgerConfig.setNewEntriesCheckDelayInMillis( + serviceConfig.getManagedLedgerNewEntriesCheckDelayInMillis()); + return managedLedgerConfig; }); } @@ -2696,7 +2779,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) { @@ -3232,5 +3315,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 daa2b249a9ca6..52ca22228a1fe 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,7 @@ import java.util.Optional; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.TimeUnit; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.atomic.AtomicInteger; import javax.annotation.Nonnull; import org.apache.commons.lang3.tuple.MutablePair; @@ -43,10 +43,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; @@ -314,7 +312,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 @@ -340,20 +338,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 7bd15992f640d..d4a695663be8e 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 @@ -136,6 +136,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 @@ -146,7 +148,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 consumer = pulsarClient.newConsumer() @@ -156,7 +159,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 75b20d521476c..bb00a35e64bf7 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 579db85dc249b..1135e2c8ff4e7 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 @@ -1096,7 +1096,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(); @@ -1140,7 +1140,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 e69230b2d2dcb..beac28cfdb16c 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; @@ -42,11 +39,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; @@ -55,7 +49,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 db717e8946cab..c9d2f78233b27 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 @@ -109,6 +109,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") @@ -193,6 +194,56 @@ public void testDurableSubscribe() throws Exception { } } + @Test(timeOut = 20000) + public void testRead() throws Exception { + Properties properties = new Properties(); + properties.setProperty("serviceUrl", brokerUrl.toString()); + properties.setProperty("useTls", "false"); + + final String topicName = getTopicWithRandomSuffix("reader"); + admin.topics().createNonPartitionedTopic(topicName); + + int numberOfMessages = 10; + @Cleanup("shutdownNow") + ExecutorService executor = Executors.newSingleThreadExecutor(); + CompletableFuture future = new CompletableFuture<>(); + executor.execute(() -> { + try { + PulsarClientTool pulsarClientToolReader = new PulsarClientTool(properties); + String[] args = {"read", "-m", "latest", "-n", Integer.toString(numberOfMessages), "--hex", "-r", "30", + topicName}; + assertEquals(pulsarClientToolReader.run(args), 0); + future.complete(null); + } catch (Throwable t) { + future.completeExceptionally(t); + } + }); + + // Make sure subscription has been created + retryStrategically((test) -> { + try { + return admin.topics().getSubscriptions(topicName).size() == 1; + } catch (Exception e) { + return false; + } + }, 10, 500); + + assertEquals(admin.topics().getSubscriptions(topicName).size(), 1); + assertTrue(admin.topics().getSubscriptions(topicName).get(0).startsWith("reader-")); + PulsarClientTool pulsarClientToolProducer = new PulsarClientTool(properties); + + String[] args = {"produce", "--messages", "Have a nice day", "-n", Integer.toString(numberOfMessages), "-r", + "20", "-p", "key1=value1", "-p", "key2=value2", "-k", "partition_key", topicName}; + assertEquals(pulsarClientToolProducer.run(args), 0); + assertFalse(future.isCompletedExceptionally()); + future.get(); + + Awaitility.await() + .ignoreExceptions() + .atMost(Duration.ofMillis(20000)) + .until(()->admin.topics().getSubscriptions(topicName).size() == 0); + } + @Test(timeOut = 20000) public void testEncryption() throws Exception { Properties properties = new Properties(); @@ -200,6 +251,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;