Skip to content

[improve][broker] Make internalGetStats to async #15348

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 3 commits into from
May 11, 2022
Merged

[improve][broker] Make internalGetStats to async #15348

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
f85b039
[improve][broker] Make internalGetStats to async
nodece Apr 28, 2022
e8d7d6e
Add test
nodece May 7, 2022
43c955c
Fix CI
nodece May 7, 2022
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
27 changes: 14 additions & 13 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1167,22 +1167,23 @@ private void internalGetSubscriptionsForNonPartitionedTopic(AsyncResponse asyncR
});
}

protected TopicStats internalGetStats(boolean authoritative, boolean getPreciseBacklog,
boolean subscriptionBacklogSize, boolean getEarliestTimeInBacklog) {
protected CompletableFuture<? extends TopicStats> internalGetStatsAsync(boolean authoritative,
boolean getPreciseBacklog,
boolean subscriptionBacklogSize,
boolean getEarliestTimeInBacklog) {
CompletableFuture<Void> future;

if (topicName.isGlobal()) {
validateGlobalNamespaceOwnership(namespaceName);
future = validateGlobalNamespaceOwnershipAsync(namespaceName);
} else {
future = CompletableFuture.completedFuture(null);
}
validateTopicOwnership(topicName, authoritative);
validateTopicOperation(topicName, TopicOperation.GET_STATS);

Topic topic = getTopicReference(topicName);
try {
return topic.asyncGetStats(getPreciseBacklog, subscriptionBacklogSize, getEarliestTimeInBacklog).get();
} catch (InterruptedException | ExecutionException e) {
log.error("[{}] Failed to get stats for {}", clientAppId(), topicName, e);
throw new RestException(Status.INTERNAL_SERVER_ERROR,
(e instanceof ExecutionException) ? e.getCause().getMessage() : e.getMessage());
}
return future.thenCompose(__ -> validateTopicOwnershipAsync(topicName, authoritative))
.thenComposeAsync(__ -> validateTopicOperationAsync(topicName, TopicOperation.GET_STATS))
.thenCompose(__ -> getTopicReferenceAsync(topicName))
.thenCompose(topic -> topic.asyncGetStats(getPreciseBacklog, subscriptionBacklogSize,
getEarliestTimeInBacklog));
}

protected PersistentTopicInternalStats internalGetInternalStats(boolean authoritative, boolean metadata) {
Expand Down
23 changes: 0 additions & 23 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/NonPersistentTopics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,15 +45,13 @@
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopic;
import org.apache.pulsar.broker.web.RestException;
import org.apache.pulsar.common.naming.Constants;
import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import org.apache.pulsar.common.policies.data.NamespaceOperation;
import org.apache.pulsar.common.policies.data.NonPersistentTopicStats;
import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.policies.data.TopicOperation;
Expand Down Expand Up @@ -89,27 +87,6 @@ public PartitionedTopicMetadata getPartitionedMetadata(@PathParam("property") St
return getPartitionedTopicMetadata(topicName, authoritative, checkAllowAutoCreation);
}

@GET
@Path("{property}/{cluster}/{namespace}/{topic}/stats")
@ApiOperation(hidden = true, value = "Get the stats for the topic.")
@ApiResponses(value = {
@ApiResponse(code = 307, message = "Current broker doesn't serve the namespace of this topic"),
@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Topic does not exist")})
public NonPersistentTopicStats getStats(@PathParam("property") String property,
@PathParam("cluster") String cluster,
@PathParam("namespace") String namespace,
@PathParam("topic") @Encoded String encodedTopic,
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
@QueryParam("getPreciseBacklog") @DefaultValue("false")
boolean getPreciseBacklog) {
validateTopicName(property, cluster, namespace, encodedTopic);
validateTopicOwnership(topicName, authoritative);
validateTopicOperation(topicName, TopicOperation.GET_STATS);
Topic topic = getTopicReference(topicName);
return ((NonPersistentTopic) topic).getStats(getPreciseBacklog, false, false);
}

Comment thread
nodece marked this conversation as resolved.
@GET
@Path("{property}/{cluster}/{namespace}/{topic}/internalStats")
@ApiOperation(hidden = true, value = "Get the internal stats for the topic.")
Expand Down
18 changes: 14 additions & 4 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v1/PersistentTopics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,6 @@
import org.apache.pulsar.common.policies.data.AuthAction;
import org.apache.pulsar.common.policies.data.PersistentOfflineTopicStats;
import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
import org.apache.pulsar.common.policies.data.TopicStats;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

Expand Down Expand Up @@ -364,13 +363,24 @@ public void getSubscriptions(@Suspended final AsyncResponse asyncResponse, @Path
@ApiResponses(value = {
@ApiResponse(code = 307, message = "Current broker doesn't serve the namespace of this topic"),
@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "Topic does not exist") })
public TopicStats getStats(@PathParam("property") String property, @PathParam("cluster") String cluster,
@ApiResponse(code = 404, message = "Topic does not exist")})
public void getStats(
@Suspended final AsyncResponse asyncResponse,
@PathParam("property") String property, @PathParam("cluster") String cluster,
@PathParam("namespace") String namespace, @PathParam("topic") @Encoded String encodedTopic,
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
@QueryParam("getPreciseBacklog") @DefaultValue("false") boolean getPreciseBacklog) {
validateTopicName(property, cluster, namespace, encodedTopic);
return internalGetStats(authoritative, getPreciseBacklog, false, false);
internalGetStatsAsync(authoritative, getPreciseBacklog, false, false)
.thenAccept(asyncResponse::resume)
.exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
Comment thread
Technoboy- marked this conversation as resolved.
if (!isRedirectException(ex)) {
log.error("[{}] Failed to get stats for {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
return null;
});
}

@GET
Expand Down
38 changes: 0 additions & 38 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/NonPersistentTopics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -50,13 +50,11 @@
import org.apache.commons.lang3.StringUtils;
import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.service.Topic;
import org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopic;
import org.apache.pulsar.broker.web.RestException;
import org.apache.pulsar.common.naming.NamespaceBundle;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
import org.apache.pulsar.common.policies.data.NamespaceOperation;
import org.apache.pulsar.common.policies.data.NonPersistentTopicStats;
import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
import org.apache.pulsar.common.policies.data.Policies;
import org.apache.pulsar.common.policies.data.TopicOperation;
Expand Down Expand Up @@ -102,42 +100,6 @@ public PartitionedTopicMetadata getPartitionedMetadata(
return super.getPartitionedMetadata(tenant, namespace, encodedTopic, authoritative, checkAllowAutoCreation);
}

@GET
@Path("{tenant}/{namespace}/{topic}/stats")
@ApiOperation(value = "Get the stats for the topic.")
@ApiResponses(value = {
@ApiResponse(code = 307, message = "Current broker doesn't serve the namespace of this topic"),
@ApiResponse(code = 401, message = "Don't have permission to manage resources on this tenant"),
@ApiResponse(code = 403, message = "Don't have admin permission"),
@ApiResponse(code = 404, message = "The tenant/namespace/topic does not exist"),
@ApiResponse(code = 412, message = "Topic name is not valid"),
@ApiResponse(code = 500, message = "Internal server error"),
})
public NonPersistentTopicStats getStats(
@ApiParam(value = "Specify the tenant", required = true)
@PathParam("tenant") String tenant,
@ApiParam(value = "Specify the namespace", required = true)
@PathParam("namespace") String namespace,
@ApiParam(value = "Specify topic name", required = true)
@PathParam("topic") @Encoded String encodedTopic,
@ApiParam(value = "Is authentication required to perform this operation")
@QueryParam("authoritative") @DefaultValue("false") boolean authoritative,
@ApiParam(value = "If return precise backlog or imprecise backlog")
@QueryParam("getPreciseBacklog") @DefaultValue("false") boolean getPreciseBacklog,
@ApiParam(value = "If return backlog size for each subscription, require locking on ledger so be careful "
+ "not to use when there's heavy traffic.")
@QueryParam("subscriptionBacklogSize") @DefaultValue("false") boolean subscriptionBacklogSize,
@ApiParam(value = "If return time of the earliest message in backlog")
@QueryParam("getEarliestTimeInBacklog") @DefaultValue("false") boolean getEarliestTimeInBacklog) {
validateTopicName(tenant, namespace, encodedTopic);
validateTopicOwnership(topicName, authoritative);
validateTopicOperation(topicName, TopicOperation.GET_STATS);

Topic topic = getTopicReference(topicName);
return ((NonPersistentTopic) topic).getStats(getPreciseBacklog, subscriptionBacklogSize,
getEarliestTimeInBacklog);
}

@GET
@Path("{tenant}/{namespace}/{topic}/internalStats")
@ApiOperation(value = "Get the internal stats for the topic.")
Expand Down
15 changes: 12 additions & 3 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/PersistentTopics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,6 @@
import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
import org.apache.pulsar.common.policies.data.SubscribeRate;
import org.apache.pulsar.common.policies.data.TopicPolicies;
import org.apache.pulsar.common.policies.data.TopicStats;
import org.apache.pulsar.common.policies.data.impl.BacklogQuotaImpl;
import org.apache.pulsar.common.policies.data.impl.DispatchRateImpl;
import org.slf4j.Logger;
Expand Down Expand Up @@ -1025,7 +1024,8 @@ public void getSubscriptions(
@ApiResponse(code = 412, message = "Topic name is not valid"),
@ApiResponse(code = 500, message = "Internal server error"),
@ApiResponse(code = 503, message = "Failed to validate global cluster configuration") })
public TopicStats getStats(
public void getStats(
@Suspended final AsyncResponse asyncResponse,
@ApiParam(value = "Specify the tenant", required = true)
@PathParam("tenant") String tenant,
@ApiParam(value = "Specify the namespace", required = true)
Expand All @@ -1042,7 +1042,16 @@ public TopicStats getStats(
@ApiParam(value = "If return time of the earliest message in backlog")
@QueryParam("getEarliestTimeInBacklog") @DefaultValue("false") boolean getEarliestTimeInBacklog) {
validateTopicName(tenant, namespace, encodedTopic);
return internalGetStats(authoritative, getPreciseBacklog, subscriptionBacklogSize, getEarliestTimeInBacklog);
internalGetStatsAsync(authoritative, getPreciseBacklog, subscriptionBacklogSize, getEarliestTimeInBacklog)
Comment thread
Technoboy- marked this conversation as resolved.
.thenAccept(asyncResponse::resume)
.exceptionally(ex -> {
// If the exception is not redirect exception we need to log it.
if (!isRedirectException(ex)) {
log.error("[{}] Failed to get stats for {}", clientAppId(), topicName, ex);
}
resumeAsyncResponseExceptionally(asyncResponse, ex);
return null;
});
}

@GET
Expand Down
62 changes: 52 additions & 10 deletions pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminTopicApiTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,38 +18,42 @@
*/
package org.apache.pulsar.broker.admin;

import static java.nio.charset.StandardCharsets.UTF_8;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertTrue;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import lombok.Cleanup;

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerConsumerBase;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.policies.data.TopicStats;
import org.apache.pulsar.common.policies.data.stats.NonPersistentTopicStatsImpl;
import org.apache.pulsar.common.policies.data.stats.TopicStatsImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;

import java.util.List;
import java.util.concurrent.TimeUnit;

import static java.nio.charset.StandardCharsets.UTF_8;

@Test(groups = "broker-admin")
public class AdminTopicApiTest extends ProducerConsumerBase {
private static final Logger log = LoggerFactory.getLogger(AdminTopicApiTest.class);

@Override
@BeforeMethod
@BeforeClass(alwaysRun = true)
protected void setup() throws Exception {
super.internalSetup();
super.producerBaseSetup();
}

@Override
@AfterMethod(alwaysRun = true)
@BeforeClass(alwaysRun = true)
protected void cleanup() throws Exception {
super.internalCleanup();
}
Expand Down Expand Up @@ -97,4 +101,42 @@ public void testPeekMessages() throws Exception {
Assert.assertEquals(new String(messages.get(3).getValue(), UTF_8), "value-3");
Assert.assertEquals(new String(messages.get(4).getValue(), UTF_8), "value-4");
}

@DataProvider
public Object[] getStatsDataProvider() {
return new Object[]{
// v1 topic
TopicDomain.persistent + "://my-property/test/my-ns/" + UUID.randomUUID(),
TopicDomain.non_persistent+ "://my-property/test/my-ns/" + UUID.randomUUID(),
//v2 topic
TopicDomain.persistent+ "://my-property/my-ns/" + UUID.randomUUID(),
TopicDomain.non_persistent+ "://my-property/my-ns/" + UUID.randomUUID(),
};
}

@Test(dataProvider = "getStatsDataProvider")
public void testGetStats(String topic) throws Exception {
admin.topics().createNonPartitionedTopic(topic);

@Cleanup
PulsarClient newPulsarClient = PulsarClient.builder()
.serviceUrl(lookupUrl.toString())
.build();

final String subscriptionName = "my-sub";
@Cleanup
Consumer<byte[]> consumer = newPulsarClient.newConsumer()
.topic(topic)
.subscriptionName(subscriptionName)
.subscribe();

TopicStats stats = admin.topics().getStats(topic);
assertNotNull(stats);
if (topic.startsWith(TopicDomain.non_persistent.value())) {
assertTrue(stats instanceof NonPersistentTopicStatsImpl);
} else {
assertTrue(stats instanceof TopicStatsImpl);
}
assertTrue(stats.getSubscriptions().containsKey(subscriptionName));
}
}
Loading