You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2022/04/07 02:07:36 UTC
[GitHub] [pulsar] congbobo184 commented on a diff in pull request #14991: [improve][transaction] Optimize topic lookup when TC end tx.
congbobo184 commented on code in PR #14991:
URL: https://github.com/apache/pulsar/pull/14991#discussion_r844572272
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferHandlerImpl.java:
##########
@@ -336,8 +291,42 @@ protected OpRequestSend newObject(Handle<OpRequestSend> handle) {
};
}
+ public CompletableFuture<ClientCnx> getClientCnxWithLookup(String topic) {
+ return pulsarClient.getConnection(topic);
+ }
+
public CompletableFuture<ClientCnx> getClientCnx(String topic) {
- return ((PulsarClientImpl) pulsarClient).getConnection(topic);
+ NamespaceService namespaceService = pulsarService.getNamespaceService();
+ CompletableFuture<NamespaceBundle> nsBundle = namespaceService.getBundleAsync(TopicName.get(topic));
+ return nsBundle
+ .thenCompose(bundle -> namespaceService.getOwnerAsync(bundle))
Review Comment:
how about use getBrokerServiceUrlAsync?
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferHandlerImpl.java:
##########
@@ -336,8 +291,42 @@ protected OpRequestSend newObject(Handle<OpRequestSend> handle) {
};
}
+ public CompletableFuture<ClientCnx> getClientCnxWithLookup(String topic) {
+ return pulsarClient.getConnection(topic);
+ }
+
public CompletableFuture<ClientCnx> getClientCnx(String topic) {
- return ((PulsarClientImpl) pulsarClient).getConnection(topic);
+ NamespaceService namespaceService = pulsarService.getNamespaceService();
+ CompletableFuture<NamespaceBundle> nsBundle = namespaceService.getBundleAsync(TopicName.get(topic));
Review Comment:
may we should think about namespace not exist
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/transaction/buffer/impl/TransactionBufferHandlerImpl.java:
##########
@@ -336,8 +291,42 @@ protected OpRequestSend newObject(Handle<OpRequestSend> handle) {
};
}
+ public CompletableFuture<ClientCnx> getClientCnxWithLookup(String topic) {
+ return pulsarClient.getConnection(topic);
+ }
+
public CompletableFuture<ClientCnx> getClientCnx(String topic) {
- return ((PulsarClientImpl) pulsarClient).getConnection(topic);
+ NamespaceService namespaceService = pulsarService.getNamespaceService();
+ CompletableFuture<NamespaceBundle> nsBundle = namespaceService.getBundleAsync(TopicName.get(topic));
+ return nsBundle
+ .thenCompose(bundle -> namespaceService.getOwnerAsync(bundle))
Review Comment:
how about use getBrokerServiceUrlAsync?
--
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