Skip to content

[fix][broker]Fix dirty reading of namespace level offload thresholds #24696

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Merged
Technoboy- merged 1 commit into from
Sep 4, 2025
Merged

[fix][broker]Fix dirty reading of namespace level offload thresholds #24696

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
68 changes: 53 additions & 15 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2182,7 +2182,7 @@ protected void internalSetOffloadThreshold(long newThreshold) {
policies.offload_policies = new OffloadPoliciesImpl();
}
((OffloadPoliciesImpl) policies.offload_policies).setManagedLedgerOffloadThresholdInBytes(newThreshold);
policies.offload_threshold = newThreshold;
mergeOffloadThresholdsForCompatibility(policies);
return policies;
});
log.info("[{}] Successfully updated offloadThreshold configuration: namespace={}, value={}",
Expand All @@ -2209,7 +2209,7 @@ protected CompletableFuture<Void> internalSetOffloadThresholdInSecondsAsync(long
}
((OffloadPoliciesImpl) policies.offload_policies)
.setManagedLedgerOffloadThresholdInSeconds(newThreshold);
policies.offload_threshold_in_seconds = newThreshold;
mergeOffloadThresholdsForCompatibility(policies);
return policies;
})
)
Expand Down Expand Up @@ -2239,7 +2239,7 @@ protected void internalSetOffloadDeletionLag(Long newDeletionLagMs) {
}
((OffloadPoliciesImpl) policies.offload_policies)
.setManagedLedgerOffloadDeletionLagInMillis(newDeletionLagMs);
policies.offload_deletion_lag_ms = newDeletionLagMs;
mergeOffloadThresholdsForCompatibility(policies);
return policies;
});
log.info("[{}] Successfully updated offloadDeletionLagMs configuration: namespace={}, value={}",
Expand Down Expand Up @@ -2352,26 +2352,59 @@ private <T> void mutatePolicy(Function<Policies, Policies> policyTransformation,
}
}

/**
* Before https://github.com/apache/pulsar/pull/6183, users can set broker level offload policies, and handle
* namespace-level thresholds by the following fields:
* - {@link Policies#offload_deletion_lag_ms}
* - {@link Policies#offload_threshold}
* - {@link Policies#offload_threshold_in_seconds}
*
* After https://github.com/apache/pulsar/pull/6183, Pulsar supports namespace-level policies, which was
* supported by {@link Policies#offload_policies}. And the thresholds were moved to the following fields:
* - {@link Policies#offload_policies} -> {@link OffloadPoliciesImpl#getManagedLedgerOffloadDeletionLagInMillis}
* - {@link Policies#offload_policies} -> {@link OffloadPoliciesImpl#getManagedLedgerOffloadThresholdInBytes}
* - {@link Policies#offload_policies} -> {@link OffloadPoliciesImpl#getManagedLedgerOffloadThresholdInSeconds}
*
* To make the offload policies compatible with the old policies, uses the old policies if the new policies
* are not set. Once the new fields are set, the old fields will be ignored.
*/
private void mergeOffloadThresholdsForCompatibility(Policies nsPolicies) {
Long oldOffloadDeletionLagMs = nsPolicies.offload_deletion_lag_ms;
Long oldOffloadThresholdInBytes = nsPolicies.offload_threshold;
Long odlOffloadThresholdInSeconds = nsPolicies.offload_threshold_in_seconds;
// If the old values are empty, skip.
if (oldOffloadDeletionLagMs == null && oldOffloadThresholdInBytes == -1 && odlOffloadThresholdInSeconds == -1) {
return;
}
// If the new values are empty, use the old values.
OffloadPoliciesImpl nsOffloadPolicies = (OffloadPoliciesImpl) nsPolicies.offload_policies;
if (Objects.equals(nsOffloadPolicies.getManagedLedgerOffloadDeletionLagInMillis(),
OffloadPoliciesImpl.DEFAULT_OFFLOAD_DELETION_LAG_IN_MILLIS) && oldOffloadDeletionLagMs != null) {
nsOffloadPolicies.setManagedLedgerOffloadDeletionLagInMillis(oldOffloadDeletionLagMs);
}
if (Objects.equals(nsOffloadPolicies.getManagedLedgerOffloadThresholdInBytes(),
OffloadPoliciesImpl.DEFAULT_OFFLOAD_THRESHOLD_IN_BYTES) && oldOffloadThresholdInBytes != -1) {
nsOffloadPolicies.setManagedLedgerOffloadThresholdInBytes(oldOffloadThresholdInBytes);
}
if (Objects.equals(nsOffloadPolicies.getManagedLedgerOffloadThresholdInSeconds(),
OffloadPoliciesImpl.DEFAULT_OFFLOAD_THRESHOLD_IN_SECONDS) && odlOffloadThresholdInSeconds != -1) {
nsOffloadPolicies.setManagedLedgerOffloadThresholdInSeconds(odlOffloadThresholdInSeconds);
}
// Since the thresholds are moved into "nsPolicies.offload_policies", remove the old fields.
nsPolicies.offload_deletion_lag_ms = null;
nsPolicies.offload_threshold = -1;
nsPolicies.offload_threshold_in_seconds = -1;
}

protected void internalSetOffloadPolicies(AsyncResponse asyncResponse, OffloadPoliciesImpl offloadPolicies) {
validateNamespacePolicyOperation(namespaceName, PolicyName.OFFLOAD, PolicyOperation.WRITE);
validatePoliciesReadOnlyAccess();
validateOffloadPolicies(offloadPolicies);

try {
namespaceResources().setPoliciesAsync(namespaceName, policies -> {
if (Objects.equals(offloadPolicies.getManagedLedgerOffloadDeletionLagInMillis(),
OffloadPoliciesImpl.DEFAULT_OFFLOAD_DELETION_LAG_IN_MILLIS)) {
offloadPolicies.setManagedLedgerOffloadDeletionLagInMillis(policies.offload_deletion_lag_ms);
} else {
policies.offload_deletion_lag_ms = offloadPolicies.getManagedLedgerOffloadDeletionLagInMillis();
}
if (Objects.equals(offloadPolicies.getManagedLedgerOffloadThresholdInBytes(),
OffloadPoliciesImpl.DEFAULT_OFFLOAD_THRESHOLD_IN_BYTES)) {
offloadPolicies.setManagedLedgerOffloadThresholdInBytes(policies.offload_threshold);
} else {
policies.offload_threshold = offloadPolicies.getManagedLedgerOffloadThresholdInBytes();
}
policies.offload_policies = offloadPolicies;
mergeOffloadThresholdsForCompatibility(policies);
return policies;
}).thenApply(r -> {
log.info("[{}] Successfully updated offload configuration: namespace={}, map={}", clientAppId(),
Expand All @@ -2397,7 +2430,12 @@ protected void internalRemoveOffloadPolicies(AsyncResponse asyncResponse) {
validatePoliciesReadOnlyAccess();
try {
namespaceResources().setPoliciesAsync(namespaceName, (policies) -> {
// Remove new offload policies.
policies.offload_policies = null;
// Remove the old offload policies thresholds.
policies.offload_deletion_lag_ms = null;
policies.offload_threshold = -1;
policies.offload_threshold_in_seconds = -1;
return policies;
}).thenApply(r -> {
log.info("[{}] Successfully remove offload configuration: namespace={}", clientAppId(), namespaceName);
Expand Down
51 changes: 51 additions & 0 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApi2Test.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,11 +125,14 @@
import org.apache.pulsar.common.policies.data.BundlesData;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.ConsumerStats;
import org.apache.pulsar.common.policies.data.DispatchRate;
import org.apache.pulsar.common.policies.data.EntryFilters;
import org.apache.pulsar.common.policies.data.FailureDomain;
import org.apache.pulsar.common.policies.data.NamespaceIsolationData;
import org.apache.pulsar.common.policies.data.NamespaceIsolationPolicyUnloadScope;
import org.apache.pulsar.common.policies.data.NonPersistentTopicStats;
import org.apache.pulsar.common.policies.data.OffloadPolicies;
import org.apache.pulsar.common.policies.data.OffloadPoliciesImpl;
import org.apache.pulsar.common.policies.data.PartitionedTopicStats;
import org.apache.pulsar.common.policies.data.PersistencePolicies;
import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
Expand Down Expand Up @@ -4041,6 +4044,54 @@ public void testGrantAndRevokePermissions() throws Exception {
Assert.assertTrue(permissions22.isEmpty());
}

@Test
public void testOverridesNamespaceOffloadThreshold() throws Exception {
String namespace = BrokerTestUtil.newUniqueName(this.defaultTenant + "/ns");
String topic = BrokerTestUtil.newUniqueName("persistent://" + namespace + "/tp");
admin.namespaces().createNamespace(namespace);
admin.topics().createNonPartitionedTopic(topic);
admin.topicPolicies().setDispatchRate(topic, DispatchRate.builder().dispatchThrottlingRateInMsg(1).build());
// assert we get -1 which indicates it will fall back to default
assertEquals(admin.namespaces().getOffloadThreshold(namespace), -1);
assertEquals(admin.namespaces().getOffloadThresholdInSeconds(namespace), -1);
// Set namespace level offloading threshold.
long m1 = 1024 * 1024;
long h1 = 1000 * 3600;
OffloadPoliciesImpl policies = OffloadPoliciesImpl.builder()
.managedLedgerOffloadDriver("S3")
.s3ManagedLedgerOffloadBucket("bucket-1")
.managedLedgerOffloadThresholdInBytes(m1)
.managedLedgerOffloadThresholdInSeconds(h1)
.build();
admin.namespaces().setOffloadPolicies(namespace, policies);
OffloadPolicies policies1 = admin.namespaces().getOffloadPolicies(namespace);
assertEquals(policies1.getManagedLedgerOffloadThresholdInBytes(), m1);
assertEquals(policies1.getManagedLedgerOffloadThresholdInSeconds(), h1);

long m2 = 2 * 1024 * 1024L;
long h2 = 2 * 1000 * 3600;
admin.namespaces().setOffloadThreshold(namespace, m2);
admin.namespaces().setOffloadThresholdInSeconds(namespace, h2);
OffloadPolicies policies2 = admin.namespaces().getOffloadPolicies(namespace);
assertEquals(policies2.getManagedLedgerOffloadThresholdInBytes(), m2);
assertEquals(policies2.getManagedLedgerOffloadThresholdInSeconds(), h2);
OffloadPolicies policies3 = admin.topicPolicies().getOffloadPolicies(topic, true);
assertEquals(policies3.getManagedLedgerOffloadThresholdInBytes(), m2);
assertEquals(policies3.getManagedLedgerOffloadThresholdInSeconds(), h2);

admin.namespaces().removeOffloadPolicies(namespace);
OffloadPolicies policies4 = admin.namespaces().getOffloadPolicies(namespace);
assertTrue(policies4 == null);
assertEquals(admin.namespaces().getOffloadThreshold(namespace), -1);
assertEquals(admin.namespaces().getOffloadThresholdInSeconds(namespace), -1);
OffloadPolicies policies5 = admin.topicPolicies().getOffloadPolicies(topic, true);
assertTrue(policies5 == null);

// cleanup.
admin.topics().delete(topic);
admin.namespaces().deleteNamespace(namespace);
}

@Test
public void testDeletePatchyPartitionedTopic() throws Exception {
final String topic = BrokerTestUtil.newUniqueName(defaultNamespace + "/tp");
Expand Down
Loading