You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "michaeljmarshall (via GitHub)" <gi...@apache.org> on 2023/01/24 19:07:15 UTC
[GitHub] [pulsar] michaeljmarshall commented on a diff in pull request #19153: [improve] PIP-241: add TopicEventListener / topic events for the BrokerService
michaeljmarshall commented on code in PR #19153:
URL: https://github.com/apache/pulsar/pull/19153#discussion_r1085749011
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java:
##########
@@ -1033,17 +1047,33 @@ public CompletableFuture<Optional<Topic>> getTopic(final TopicName topicName, bo
});
} else {
return topics.computeIfAbsent(topicName.toString(), (name) -> {
+ topicEventsDispatcher.notify(topicName.toString(), TopicEvent.LOAD, EventStage.BEFORE);
if (topicName.isPartitioned()) {
final TopicName partitionedTopicName = TopicName.get(topicName.getPartitionedTopicName());
return this.fetchPartitionedTopicMetadataAsync(partitionedTopicName).thenCompose((metadata) -> {
if (topicName.getPartitionIndex() < metadata.partitions) {
- return createNonPersistentTopic(name);
+ topicEventsDispatcher
+ .notify(topicName.toString(), TopicEvent.CREATE, EventStage.BEFORE);
+
+ CompletableFuture<Optional<Topic>> res = createNonPersistentTopic(name);
+
+ topicEventsDispatcher.notifyOnCompletion(res, topicName.toString(), TopicEvent.CREATE);
+ topicEventsDispatcher.notifyOnCompletion(res, topicName.toString(), TopicEvent.LOAD);
+ return res;
}
+ topicEventsDispatcher.notify(topicName.toString(), TopicEvent.LOAD, EventStage.FAILURE);
return CompletableFuture.completedFuture(Optional.empty());
});
} else if (createIfMissing) {
- return createNonPersistentTopic(name);
+ topicEventsDispatcher.notify(name, TopicEvent.CREATE, EventStage.BEFORE);
+
+ CompletableFuture<Optional<Topic>> res = createNonPersistentTopic(name);
+
+ topicEventsDispatcher.notifyOnCompletion(res, name, TopicEvent.CREATE);
+ topicEventsDispatcher.notifyOnCompletion(res, name, TopicEvent.LOAD);
Review Comment:
It looks like `CREATE` always implies `LOAD` in this PR. Do we expect this to change? I wonder if we should declare in the design that `CREATE` always implies `LOAD` and then we can skip the extra notification because implementations know that `CREATE` also means `LOAD`. Obviously, we still need `LOAD` because `LOAD` does not imply `CREATE`. (This is not a strong suggestion, just wondering about messages and the contract we're creating.)
Thinking about this further, I see that on these lines, we're completing `LOAD` and `CREATE` events that we already started. Perhaps collapsing the events would make the design too confusing.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java:
##########
@@ -1033,17 +1047,33 @@ public CompletableFuture<Optional<Topic>> getTopic(final TopicName topicName, bo
});
} else {
return topics.computeIfAbsent(topicName.toString(), (name) -> {
+ topicEventsDispatcher.notify(topicName.toString(), TopicEvent.LOAD, EventStage.BEFORE);
if (topicName.isPartitioned()) {
final TopicName partitionedTopicName = TopicName.get(topicName.getPartitionedTopicName());
return this.fetchPartitionedTopicMetadataAsync(partitionedTopicName).thenCompose((metadata) -> {
if (topicName.getPartitionIndex() < metadata.partitions) {
- return createNonPersistentTopic(name);
+ topicEventsDispatcher
+ .notify(topicName.toString(), TopicEvent.CREATE, EventStage.BEFORE);
+
+ CompletableFuture<Optional<Topic>> res = createNonPersistentTopic(name);
+
+ topicEventsDispatcher.notifyOnCompletion(res, topicName.toString(), TopicEvent.CREATE);
+ topicEventsDispatcher.notifyOnCompletion(res, topicName.toString(), TopicEvent.LOAD);
Review Comment:
Nit: do we care about the order of these notifications? Technically, the `LOAD` will come before the `CREATE` in cases where the `res` is not yet completed because future callbacks are tracked with a stack and are completed in reverse order.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java:
##########
@@ -2153,7 +2211,9 @@ private CompletableFuture<Void> removeTopicFutureFromCache(String topic,
TopicName topicName = TopicName.get(topic);
return pulsar.getNamespaceService().getBundleAsync(topicName)
.thenAccept(namespaceBundle -> {
+ topicEventsDispatcher.notify(topic, TopicEvent.UNLOAD, EventStage.BEFORE);
removeTopicFromCache(topic, namespaceBundle, createTopicFuture);
+ topicEventsDispatcher.notify(topic, TopicEvent.UNLOAD, EventStage.SUCCESS);
Review Comment:
Nit: why not add this notification to `removeTopicFromCache`? Then we could collapse the duplicated code in this method and in `cleanUnloadedTopicFromCache`.
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java:
##########
@@ -1064,6 +1094,13 @@ public CompletableFuture<Optional<Topic>> getTopic(final TopicName topicName, bo
}
public CompletableFuture<Void> deleteTopic(String topic, boolean forceDelete) {
+ topicEventsDispatcher.notify(topic, TopicEvent.DELETE, EventStage.BEFORE);
+ CompletableFuture<Void> result = deleteTopicInternal(topic, forceDelete);
+ topicEventsDispatcher.notifyOnCompletion(result, topic, TopicEvent.DELETE);
Review Comment:
Analogously to `CREATE`/`LOAD`, it looks like we do not notify `UNLOAD` when a topic is deleted. I haven't looked at the internals of the `deteTopicInternal` method, so correct me if I am wrong.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org