Skip to content
New issue

Have a question about this project? # for a free GitHub account to open an issue and contact its maintainers and the community.

#

By clicking “#”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? # to your account

[improve][broker] Do not retain the data in the system topic #22022

Merged
merged 5 commits into from
Feb 6, 2024
Merged

[improve][broker] Do not retain the data in the system topic #22022

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
909d569
[improve][broker] Do not retain the data in the system topic
liangyepianzhou Feb 4, 2024
546ddd8
test: testRetentionPolicesForSystemTopic
liangyepianzhou Feb 4, 2024
ea4248c
log: add a debug log
liangyepianzhou Feb 5, 2024
848ceea
Update pulsar-broker/src/main/java/org/apache/pulsar/broker/service/B…
liangyepianzhou Feb 5, 2024
91c6ca8
revert: Delete the modification for negative retention time.
liangyepianzhou Feb 5, 2024
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
15 changes: 11 additions & 4 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1779,10 +1779,17 @@ private CompletableFuture<ManagedLedgerConfig> getManagedLedgerConfig(@Nonnull T
}

if (retentionPolicies == null) {
retentionPolicies = policies.map(p -> p.retention_policies).orElseGet(
() -> new RetentionPolicies(serviceConfig.getDefaultRetentionTimeInMinutes(),
serviceConfig.getDefaultRetentionSizeInMB())
);
if (SystemTopicNames.isSystemTopic(topicName)) {
if (log.isDebugEnabled()) {
log.debug("{} Disable data retention policy for system topic.", topicName);
}
retentionPolicies = new RetentionPolicies(0, 0);
} else {
retentionPolicies = policies.map(p -> p.retention_policies).orElseGet(
() -> new RetentionPolicies(serviceConfig.getDefaultRetentionTimeInMinutes(),
serviceConfig.getDefaultRetentionSizeInMB())
);
}
}

ManagedLedgerConfig managedLedgerConfig = new ManagedLedgerConfig();
Expand Down
48 changes: 48 additions & 0 deletions pulsar-broker/src/test/java/org/apache/pulsar/compaction/CompactionRetentionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,16 +38,21 @@
import lombok.Cleanup;
import lombok.extern.slf4j.Slf4j;
import org.apache.bookkeeper.client.BookKeeper;
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Reader;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.common.naming.SystemTopicNames;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.RetentionPolicies;
import org.apache.pulsar.common.policies.data.TenantInfoImpl;
import org.awaitility.Awaitility;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;
Expand Down Expand Up @@ -212,6 +217,49 @@ public void testCompactionRetentionOnTopicCreationWithTopicPolicies() throws Exc
);
}

@Test
public void testRetentionPolicesForSystemTopic() throws Exception {
String namespace = "my-tenant/my-ns";
String topicPrefix = "persistent://" + namespace + "/";
admin.namespaces().setRetention(namespace, new RetentionPolicies(-1, -1));
// Check event topics and transaction internal topics.
for (String eventTopic : SystemTopicNames.EVENTS_TOPIC_NAMES) {
checkSystemTopicRetentionPolicy(topicPrefix + eventTopic);
}
checkSystemTopicRetentionPolicy(topicPrefix + SystemTopicNames.TRANSACTION_COORDINATOR_ASSIGN);
checkSystemTopicRetentionPolicy(topicPrefix + SystemTopicNames.TRANSACTION_COORDINATOR_LOG);
checkSystemTopicRetentionPolicy(topicPrefix + SystemTopicNames.PENDING_ACK_STORE_SUFFIX);

// Check common topics.
checkCommonTopicRetentionPolicy(topicPrefix + "my-topic" + System.nanoTime());
// Specify retention policies for system topic.
pulsar.getConfiguration().setTopicLevelPoliciesEnabled(true);
pulsar.getConfiguration().setSystemTopicEnabled(true);
admin.topics().createNonPartitionedTopic(topicPrefix + SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT);
admin.topicPolicies().setRetention(topicPrefix + SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT,
new RetentionPolicies(10, 10));
Awaitility.await().untilAsserted(() -> {
checkTopicRetentionPolicy(topicPrefix + SystemTopicNames.TRANSACTION_BUFFER_SNAPSHOT,
new RetentionPolicies(10, 10));
});
}

private void checkSystemTopicRetentionPolicy(String topicName) throws Exception {
checkTopicRetentionPolicy(topicName, new RetentionPolicies(0, 0));

}

private void checkCommonTopicRetentionPolicy(String topicName) throws Exception {
checkTopicRetentionPolicy(topicName, new RetentionPolicies(-1, -1));
}

private void checkTopicRetentionPolicy(String topicName, RetentionPolicies retentionPolicies) throws Exception {
ManagedLedgerConfig config = pulsar.getBrokerService()
.getManagedLedgerConfig(TopicName.get(topicName)).get();
Assert.assertEquals(config.getRetentionSizeInMB(), retentionPolicies.getRetentionSizeInMB());
Assert.assertEquals(config.getRetentionTimeMillis(),retentionPolicies.getRetentionTimeInMinutes() * 60000L);
}

private void testCompactionCursorRetention(String topic) throws Exception {
Set<String> keys = Sets.newHashSet("a", "b", "c");
Set<String> keysToExpire = Sets.newHashSet("x1", "x2");
Expand Down
Loading