This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-3.2
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-3.2 by this push:
new c3c17dee756 [fix][broker] avoid offload system topic (#22497)
c3c17dee756 is described below
commit c3c17dee7567d0a182affb1991e1e35098689d9b
Author: Qiang Zhao <mattisonc...@apache.org>
AuthorDate: Wed May 8 13:10:49 2024 +0800
[fix][broker] avoid offload system topic (#22497)
Co-authored-by: 道君 <dao...@apache.org>
---
.../pulsar/broker/service/BrokerService.java | 8 +-
.../pulsar/broker/service/BrokerServiceTest.java | 94 ++++++++++++++++++++++
2 files changed, 101 insertions(+), 1 deletion(-)
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 032d4dd9369..60d56c0d908 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
@@ -1906,7 +1906,13 @@ public class BrokerService implements Closeable {
topicLevelOffloadPolicies,
OffloadPoliciesImpl.oldPoliciesCompatible(nsLevelOffloadPolicies,
policies.orElse(null)),
getPulsar().getConfig().getProperties());
- if
(NamespaceService.isSystemServiceNamespace(namespace.toString())) {
+ if (NamespaceService.isSystemServiceNamespace(namespace.toString())
+ || SystemTopicNames.isSystemTopic(topicName)) {
+ /*
+ Avoid setting broker internal system topics using off-loader
because some of them are the
+ preconditions of other topics. The slow replying log speed
will cause a delay in all the topic
+ loading.(timeout)
+ */
managedLedgerConfig.setLedgerOffloader(NullLedgerOffloader.INSTANCE);
} else {
if (topicLevelOffloadPolicies != null) {
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 fcf11fad708..ab0b8f813ea 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
@@ -67,12 +67,15 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.client.api.ReadHandle;
+import org.apache.bookkeeper.mledger.LedgerOffloader;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
import org.apache.bookkeeper.mledger.ManagedLedgerException;
import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerFactoryImpl;
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
import org.apache.commons.lang3.StringUtils;
+import org.apache.bookkeeper.mledger.impl.NullLedgerOffloader;
import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
@@ -111,6 +114,9 @@ import org.apache.pulsar.common.naming.SystemTopicNames;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.BundlesData;
import org.apache.pulsar.common.policies.data.LocalPolicies;
+import org.apache.pulsar.common.policies.data.OffloadPolicies;
+import org.apache.pulsar.common.policies.data.OffloadPoliciesImpl;
+import org.apache.pulsar.common.policies.data.OffloadedReadPriority;
import org.apache.pulsar.common.policies.data.SubscriptionStats;
import org.apache.pulsar.common.policies.data.TopicStats;
import org.apache.pulsar.common.protocol.Commands;
@@ -1772,4 +1778,92 @@ public class BrokerServiceTest extends BrokerTestBase {
fail("Unsubscribe failed");
}
}
+
+
+ @Test
+ public void testOffloadConfShouldNotAppliedForSystemTopic() throws
PulsarAdminException {
+ final String driver = "aws-s3";
+ final String region = "test-region";
+ final String bucket = "test-bucket";
+ final String role = "test-role";
+ final String roleSessionName = "test-role-session-name";
+ final String credentialId = "test-credential-id";
+ final String credentialSecret = "test-credential-secret";
+ final String endPoint = "test-endpoint";
+ final Integer maxBlockSizeInBytes = 5;
+ final Integer readBufferSizeInBytes = 2;
+ final Long offloadThresholdInBytes = 10L;
+ final Long offloadThresholdInSeconds = 1000L;
+ final Long offloadDeletionLagInMillis = 5L;
+
+ final OffloadPoliciesImpl offloadPolicies = OffloadPoliciesImpl.create(
+ driver,
+ region,
+ bucket,
+ endPoint,
+ role,
+ roleSessionName,
+ credentialId,
+ credentialSecret,
+ maxBlockSizeInBytes,
+ readBufferSizeInBytes,
+ offloadThresholdInBytes,
+ offloadThresholdInSeconds,
+ offloadDeletionLagInMillis,
+ OffloadedReadPriority.TIERED_STORAGE_FIRST
+ );
+
+ var fakeOffloader = new LedgerOffloader() {
+ @Override
+ public String getOffloadDriverName() {
+ return driver;
+ }
+
+ @Override
+ public CompletableFuture<Void> offload(ReadHandle ledger, UUID
uid, Map<String, String> extraMetadata) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<ReadHandle> readOffloaded(long ledgerId,
UUID uid, Map<String, String> offloadDriverMetadata) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public CompletableFuture<Void> deleteOffloaded(long ledgerId, UUID
uid, Map<String, String> offloadDriverMetadata) {
+ return CompletableFuture.completedFuture(null);
+ }
+
+ @Override
+ public OffloadPolicies getOffloadPolicies() {
+ return offloadPolicies;
+ }
+
+ @Override
+ public void close() {
+ }
+ };
+
+ final BrokerService brokerService = pulsar.getBrokerService();
+ final String namespace = "prop/" + UUID.randomUUID();
+ admin.namespaces().createNamespace(namespace);
+ admin.namespaces().setOffloadPolicies(namespace, offloadPolicies);
+
+ // Inject the cache to avoid real load off-loader jar
+ final Map<NamespaceName, LedgerOffloader> ledgerOffloaderMap =
pulsar.getLedgerOffloaderMap();
+ ledgerOffloaderMap.put(NamespaceName.get(namespace), fakeOffloader);
+
+ // (1) test normal topic
+ final String normalTopic = "persistent://" + namespace + "/" +
UUID.randomUUID();
+ var managedLedgerConfig =
brokerService.getManagedLedgerConfig(TopicName.get(normalTopic)).join();
+
+ Assert.assertEquals(managedLedgerConfig.getLedgerOffloader(),
fakeOffloader);
+
+ // (2) test system topic
+ for (String eventTopicName : SystemTopicNames.EVENTS_TOPIC_NAMES) {
+ managedLedgerConfig =
brokerService.getManagedLedgerConfig(TopicName.get(eventTopicName)).join();
+ Assert.assertEquals(managedLedgerConfig.getLedgerOffloader(),
NullLedgerOffloader.INSTANCE);
+ }
+ }
}
+
