Skip to content

Commit

Permalink
[fix][admin] Fix namespace admin api exception response (#22587)
Browse files Browse the repository at this point in the history
  • Loading branch information
@coderzc
coderzc committed Apr 26, 2024
1 parent 3de14c5 commit f25776d
Show file tree
Hide file tree
Showing 2 changed files with 48 additions and 17 deletions.
5 changes: 3 additions & 2 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -2019,7 +2019,7 @@ protected void internalSetMaxUnackedMessagesPerConsumer(Integer maxUnackedMessag
}

protected void internalSetMaxSubscriptionsPerTopic(Integer maxSubscriptionsPerTopic){
validateNamespacePolicyOperationAsync(namespaceName, PolicyName.MAX_SUBSCRIPTIONS, PolicyOperation.WRITE);
validateNamespacePolicyOperation(namespaceName, PolicyName.MAX_SUBSCRIPTIONS, PolicyOperation.WRITE);
validatePoliciesReadOnlyAccess();
if (maxSubscriptionsPerTopic != null && maxSubscriptionsPerTopic < 0) {
throw new RestException(Status.PRECONDITION_FAILED,
Expand Down Expand Up @@ -2125,9 +2125,10 @@ protected CompletableFuture<Void> internalSetOffloadThresholdInSecondsAsync(long
f.complete(null);
})
.exceptionally(t -> {
Throwable cause = FutureUtil.unwrapCompletionException(t);
log.error("[{}] Failed to update offloadThresholdInSeconds configuration for namespace {}",
clientAppId(), namespaceName, t);
f.completeExceptionally(new RestException(t));
f.completeExceptionally(new RestException(cause));
return null;
});

Expand Down
60 changes: 45 additions & 15 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespaceAuthZTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@

package org.apache.pulsar.broker.admin;

import static org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest.deleteNamespaceWithRetry;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNull;
import static org.testng.Assert.assertTrue;
Expand Down Expand Up @@ -58,7 +59,6 @@
import org.testng.annotations.AfterClass;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

@Test(groups = "broker-admin")
Expand All @@ -72,8 +72,6 @@ public class NamespaceAuthZTest extends MockedPulsarStandalone {

private AuthorizationService authorizationService;

private AuthorizationService orignalAuthorizationService;

private static final String TENANT_ADMIN_SUBJECT = UUID.randomUUID().toString();
private static final String TENANT_ADMIN_TOKEN = Jwts.builder()
.claim("sub", TENANT_ADMIN_SUBJECT).signWith(SECRET_KEY).compact();
Expand All @@ -100,6 +98,9 @@ public void setup() {
.authentication(new AuthenticationToken(TENANT_ADMIN_TOKEN))
.build();
this.pulsarClient = super.getPulsarService().getClient();
this.authorizationService = Mockito.spy(getPulsarService().getBrokerService().getAuthorizationService());
FieldUtils.writeField(getPulsarService().getBrokerService(), "authorizationService",
authorizationService, true);
}


Expand All @@ -115,19 +116,9 @@ public void cleanup() {
close();
}

@BeforeMethod
public void before() throws IllegalAccessException {
orignalAuthorizationService = getPulsarService().getBrokerService().getAuthorizationService();
authorizationService = Mockito.spy(orignalAuthorizationService);
FieldUtils.writeField(getPulsarService().getBrokerService(), "authorizationService",
authorizationService, true);
}

@AfterMethod
public void after() throws IllegalAccessException, PulsarAdminException {
FieldUtils.writeField(getPulsarService().getBrokerService(), "authorizationService",
orignalAuthorizationService, true);
superUserAdmin.namespaces().deleteNamespace("public/default", true);
public void after() throws Exception {
deleteNamespaceWithRetry("public/default", true, superUserAdmin);
superUserAdmin.namespaces().createNamespace("public/default");
}

Expand Down Expand Up @@ -1028,4 +1019,43 @@ public void testPackageAPI() throws Exception {
superUserAdmin.namespaces().revokePermissionsOnNamespace(namespace, subject);
}
}

@Test
@SneakyThrows
public void testOffloadThresholdInSeconds() {
final String namespace = "public/default";
final String subject = UUID.randomUUID().toString();
final String token = Jwts.builder()
.claim("sub", subject).signWith(SECRET_KEY).compact();
@Cleanup final PulsarAdmin subAdmin = PulsarAdmin.builder()
.serviceHttpUrl(getPulsarService().getWebServiceAddress())
.authentication(new AuthenticationToken(token))
.build();
Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class,
() -> subAdmin.namespaces().getOffloadThresholdInSeconds(namespace));

Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class,
() -> subAdmin.namespaces().setOffloadThresholdInSeconds(namespace, 10000));
}

@Test
@SneakyThrows
public void testMaxSubscriptionsPerTopic() {
final String namespace = "public/default";
final String subject = UUID.randomUUID().toString();
final String token = Jwts.builder()
.claim("sub", subject).signWith(SECRET_KEY).compact();
@Cleanup final PulsarAdmin subAdmin = PulsarAdmin.builder()
.serviceHttpUrl(getPulsarService().getWebServiceAddress())
.authentication(new AuthenticationToken(token))
.build();
Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class,
() -> subAdmin.namespaces().getMaxSubscriptionsPerTopic(namespace));

Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class,
() -> subAdmin.namespaces().setMaxSubscriptionsPerTopic(namespace, 100));

Assert.assertThrows(PulsarAdminException.NotAuthorizedException.class,
() -> subAdmin.namespaces().removeMaxSubscriptionsPerTopic(namespace));
}
}

0 comments on commit f25776d

Please sign in to comment.