Skip to content

[PIP-82] [pulsar-broker] incorporate review feedback #10201

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
jerrypeng merged 1 commit into from
Apr 17, 2021
Merged

[PIP-82] [pulsar-broker] incorporate review feedback #10201

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
2 changes: 1 addition & 1 deletion pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -235,7 +235,7 @@ flexible messaging model and an intuitive client API.</description>
<errorprone.version>2.5.1</errorprone.version>
<errorprone.javac.version>9+181-r4173-1</errorprone.javac.version>
<errorprone-slf4j.version>0.1.4</errorprone-slf4j.version>
<lightproto-maven-plugin.version>0.2</lightproto-maven-plugin.version>
<lightproto-maven-plugin.version>0.4</lightproto-maven-plugin.version>

<!-- Used to configure rename.netty.native. Libs -->
<rename.netty.native.libs>rename-netty-native-libs.sh</rename.netty.native.libs>
Expand Down
6 changes: 0 additions & 6 deletions pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -687,12 +687,6 @@ public class ServiceConfiguration implements PulsarConfiguration {
)
private String resourceUsageTransportClassName = "";

@FieldContext(

@jerrypeng jerrypeng Apr 12, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What was the rationale for making this not configurable?

@bharanic-dev bharanic-dev Apr 12, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@jerrypeng while reviewing the previous PR(#10008), @codelipenghui and @315157973 gave feedback that the topic name is internal implementation detail that the user doesn't need to be exposed to. I did not have a strong reason to go the other way, so made this change. We can always make it configurable in the future, if we find that there is a use-case for it.

category = CATEGORY_POLICIES,
doc = "Topic to publish usage reports to if resourceUsagePublishToTopic is enabled."
)
private String resourceUsageTransportPublishTopicName = "non-persistent://pulsar/system/resource-usage";

@FieldContext(
dynamic = true,
category = CATEGORY_POLICIES,
Expand Down
2 changes: 2 additions & 0 deletions pulsar-broker/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -491,6 +491,8 @@
<version>${lightproto-maven-plugin.version}</version>
<configuration>
<sources>${project.basedir}/src/main/proto/ResourceUsage.proto</sources>
<targetSourcesSubDir>generated-sources/lightproto/java</targetSourcesSubDir>
<targetTestSourcesSubDir>generated-sources/lightproto/java</targetTestSourcesSubDir>
</configuration>
<executions>
<execution>
Expand Down
5 changes: 5 additions & 0 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -329,6 +329,11 @@ public CompletableFuture<Void> closeAsync() {
}

// close the service in reverse order v.s. in which they are started
if (this.resourceUsageTransportManager != null) {
this.resourceUsageTransportManager.close();
this.resourceUsageTransportManager = null;
}

if (this.webService != null) {
try {
this.webService.close();
Expand Down
44 changes: 36 additions & 8 deletions ...r/src/main/java/org/apache/pulsar/broker/resourcegroup/ResourceUsageTransportManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,7 +66,7 @@ private Producer<byte[]> createProducer() throws PulsarClientException {
final int sendTimeoutSecs = 10;

return pulsarClient.newProducer()
.topic(pulsarService.getConfig().getResourceUsageTransportPublishTopicName())
.topic(RESOURCE_USAGE_TOPIC_NAME)
.batchingMaxPublishDelay(publishDelayMilliSecs, TimeUnit.MILLISECONDS)
.sendTimeout(sendTimeoutSecs, TimeUnit.SECONDS)
.blockIfQueueFull(false)
Expand Down Expand Up @@ -113,11 +113,12 @@ public void close() throws Exception {

private class ResourceUsageReader implements ReaderListener<byte[]>, AutoCloseable {
private final ResourceUsageInfo recdUsageInfo = new ResourceUsageInfo();

private final Reader<byte[]> consumer;

public ResourceUsageReader() throws PulsarClientException {
consumer = pulsarClient.newReader()
.topic(pulsarService.getConfig().getResourceUsageTransportPublishTopicName())
.topic(RESOURCE_USAGE_TOPIC_NAME)
.startMessageId(MessageId.latest)
.readerListener(this)
.create();
Expand All @@ -130,9 +131,19 @@ public void close() throws Exception {

@Override
public void received(Reader<byte[]> reader, Message<byte[]> msg) {
try {
recdUsageInfo.parseFrom(Unpooled.wrappedBuffer(msg.getData()), msg.getData().length);
long publishTime = msg.getPublishTime();
long currentTime = System.currentTimeMillis();
long timeDelta = currentTime - publishTime;

recdUsageInfo.parseFrom(Unpooled.wrappedBuffer(msg.getData()), msg.getData().length);
if (timeDelta > TimeUnit.SECONDS.toMillis(
2 * pulsarService.getConfig().getResourceUsageTransportPublishIntervalInSecs())) {
LOG.error("Stale resource usage msg from broker {} publish time {} current time{}",
recdUsageInfo.getBroker(), publishTime, currentTime);
staleMessageCount++;
return;
}
try {
recdUsageInfo.getUsageMapsList().forEach(ru -> {
ResourceUsageConsumer owner = consumerMap.get(ru.getOwner());
if (owner != null) {
Expand All @@ -150,6 +161,7 @@ public void received(Reader<byte[]> reader, Message<byte[]> msg) {
}

private static final Logger LOG = LoggerFactory.getLogger(ResourceUsageTransportManager.class);
public static final String RESOURCE_USAGE_TOPIC_NAME = "non-persistent://pulsar/system/resource-usage";
private final PulsarService pulsarService;
private final PulsarClient pulsarClient;
private final ResourceUsageWriterTask pTask;
Expand All @@ -159,9 +171,11 @@ public void received(Reader<byte[]> reader, Message<byte[]> msg) {
private final Map<String, ResourceUsageConsumer>
consumerMap = new ConcurrentHashMap<String, ResourceUsageConsumer>();

private long staleMessageCount = 0;

@jerrypeng jerrypeng Apr 12, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This variable doesn't seem to be used anywhere?

@bharanic-dev bharanic-dev Apr 12, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I added the variable with the intention of adding a unit test for it. But could not figure out a easy way to do so. I am open to ideas if you have suggestions on how I can add one. Else, I will remove this variable.


private void createTenantAndNamespace() throws PulsarServerException, PulsarAdminException {
// Create a public tenant and default namespace
TopicName topicName = TopicName.get(pulsarService.getConfig().getResourceUsageTransportPublishTopicName());
TopicName topicName = TopicName.get(RESOURCE_USAGE_TOPIC_NAME);

PulsarAdmin admin = pulsarService.getAdminClient();
ServiceConfiguration config = pulsarService.getConfig();
Expand All @@ -172,12 +186,26 @@ private void createTenantAndNamespace() throws PulsarServerException, PulsarAdmi

List<String> tenantList = admin.tenants().getTenants();
if (!tenantList.contains(tenant)) {
admin.tenants().createTenant(tenant,
new TenantInfo(Sets.newHashSet(config.getSuperUserRoles()), Sets.newHashSet(cluster)));
try {

@Hongten Hongten Feb 10, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

To add try/catch is a good solution.

admin.tenants().createTenant(tenant,
new TenantInfo(Sets.newHashSet(config.getSuperUserRoles()), Sets.newHashSet(cluster)));
} catch (PulsarAdminException ex1) {
if (!(ex1 instanceof PulsarAdminException.ConflictException)) {
LOG.error("Unexpected exception {} when creating tenant {}", ex1, tenant);
throw ex1;
}
}
}
List<String> nsList = admin.namespaces().getNamespaces(tenant);
if (!nsList.contains(namespace)) {
admin.namespaces().createNamespace(namespace);
try {
admin.namespaces().createNamespace(namespace);
} catch (PulsarAdminException ex1) {
if (!(ex1 instanceof PulsarAdminException.ConflictException)) {
LOG.error("Unexpected exception {} when creating namespace {}", ex1, namespace);
throw ex1;
}
}
}
}

Expand Down
5 changes: 1 addition & 4 deletions ...c/test/java/org/apache/pulsar/broker/resourcegroup/ResourceUsageTransportManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,7 +34,6 @@

public class ResourceUsageTransportManagerTest extends MockedPulsarServiceBaseTest {

private static final String INTERNAL_TOPIC = "non-persistent://pulsar-test/test/resource-usage";
private static final int PUBLISH_INTERVAL_SECS = 1;

@BeforeClass
Expand All @@ -53,11 +52,10 @@ protected void cleanup() throws Exception {
@Test
public void testNamespaceCreation() throws Exception {
ResourceUsageTransportManager tManager = new ResourceUsageTransportManager(pulsar);
TopicName topicName = TopicName.get(INTERNAL_TOPIC);
TopicName topicName = TopicName.get(ResourceUsageTransportManager.RESOURCE_USAGE_TOPIC_NAME);

assertTrue(admin.tenants().getTenants().contains(topicName.getTenant()));
assertTrue(admin.namespaces().getNamespaces(topicName.getTenant()).contains(topicName.getNamespace()));

}

@Test
Expand Down Expand Up @@ -116,7 +114,6 @@ public void acceptResourceUsage(String broker, ResourceUsage resourceUsage) {

private void prepareData() throws PulsarAdminException {
this.conf.setResourceUsageTransportClassName("org.apache.pulsar.broker.resourcegroup.ResourceUsageTransportManager");
this.conf.setResourceUsageTransportPublishTopicName(INTERNAL_TOPIC);
this.conf.setResourceUsageTransportPublishIntervalInSecs(PUBLISH_INTERVAL_SECS);
admin.clusters().createCluster("test", new ClusterData(pulsar.getBrokerServiceUrl()));
}
Expand Down