Re: [PR] [improve] PIP-307: Use assigned broker URL hints during broker reconnection [pulsar-client-go]
BewareMyPower merged PR #1208: URL: https://github.com/apache/pulsar-client-go/pull/1208 -- 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-unsubscr...@pulsar.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [improve] PIP-307: Use assigned broker URL hints during broker reconnection [pulsar-client-go]
BewareMyPower commented on code in PR #1208:
URL: https://github.com/apache/pulsar-client-go/pull/1208#discussion_r1576550511
##
pulsar/producer_partition.go:
##
@@ -429,12 +453,22 @@ func (p *partitionProducer) reconnectToBroker() {
return
}
- if p.options.BackoffPolicy == nil {
+ var assignedBrokerURL string
+
+ if connectionClosed != nil && connectionClosed.HasURL() {
+ delayReconnectTime = 0
+ assignedBrokerURL = connectionClosed.assignedBrokerURL
+ connectionClosed = nil // Only attempt once
Review Comment:
I see, when the client failed to connect the assigned broker URL, it needs
to call `grabCnx("")` for the lookup.
It makes sense to me. Thanks!
--
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-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] [improve] PIP-307: Use assigned broker URL hints during broker reconnection [pulsar-client-go]
dragosvictor commented on code in PR #1208:
URL: https://github.com/apache/pulsar-client-go/pull/1208#discussion_r1576544492
##
pulsar/producer_partition.go:
##
@@ -429,12 +453,22 @@ func (p *partitionProducer) reconnectToBroker() {
return
}
- if p.options.BackoffPolicy == nil {
+ var assignedBrokerURL string
+
+ if connectionClosed != nil && connectionClosed.HasURL() {
+ delayReconnectTime = 0
+ assignedBrokerURL = connectionClosed.assignedBrokerURL
+ connectionClosed = nil // Only attempt once
Review Comment:
`assignedBrokerUrl` being set to empty is what signals to `grabCnx` to
execute the lookup further on
([ref](https://github.com/apache/pulsar-client-go/pull/1208/files#diff-8adb09af2175be5751840343e14df2220a2154ca4bc8de21157ed112be25b6b9R224-R239)),
so it cannot be fixed outside of the loop and never changed subsequently.
--
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-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] [improve] PIP-307: Use assigned broker URL hints during broker reconnection [pulsar-client-go]
BewareMyPower commented on code in PR #1208:
URL: https://github.com/apache/pulsar-client-go/pull/1208#discussion_r1576526236
##
pulsar/producer_partition.go:
##
@@ -429,12 +453,22 @@ func (p *partitionProducer) reconnectToBroker() {
return
}
- if p.options.BackoffPolicy == nil {
+ var assignedBrokerURL string
+
+ if connectionClosed != nil && connectionClosed.HasURL() {
+ delayReconnectTime = 0
+ assignedBrokerURL = connectionClosed.assignedBrokerURL
+ connectionClosed = nil // Only attempt once
Review Comment:
How about moving the `assignedBrokerURL` out of the for loop?
```go
assignedBrokerURL = ""
if connectionClosed != nil && connectionClosed.HasURL() {
assignedBrokerURL = connectionClosed.assignedBrokerURL
}
for maxRetry != 0 {
// ...
if p.options.BackoffPolicy == nil {
delayReconnectTime = defaultBackoff.Next()
} else {
delayReconnectTime = p.options.BackoffPolicy.Next()
}
// ...
err := p.grabCnx(assignedBrokerURL)
```
When `connectionClosed` is nil, `assignedBrokerURL` will be an empty string
and `grabCnx(assignedBrokerURL)` will be equivalent with `grabCnx("")`.
The code will look more clear and less code changes are needed.
--
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-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] [improve] PIP-307: Use assigned broker URL hints during broker reconnection [pulsar-client-go]
BewareMyPower commented on code in PR #1208:
URL: https://github.com/apache/pulsar-client-go/pull/1208#discussion_r1576526236
##
pulsar/producer_partition.go:
##
@@ -429,12 +453,22 @@ func (p *partitionProducer) reconnectToBroker() {
return
}
- if p.options.BackoffPolicy == nil {
+ var assignedBrokerURL string
+
+ if connectionClosed != nil && connectionClosed.HasURL() {
+ delayReconnectTime = 0
+ assignedBrokerURL = connectionClosed.assignedBrokerURL
+ connectionClosed = nil // Only attempt once
Review Comment:
How about moving the `assignedBrokerURL` out of the for loop?
```go
var assignedBrokerURL string
if connectionClosed != nil && connectionClosed.HasURL() {
assignedBrokerURL = connectionClosed.assignedBrokerURL
}
for maxRetry != 0 {
// ...
if p.options.BackoffPolicy == nil {
delayReconnectTime = defaultBackoff.Next()
} else {
delayReconnectTime = p.options.BackoffPolicy.Next()
}
// ...
err := p.grabCnx(assignedBrokerURL)
```
When `connectionClosed` is nil, `assignedBrokerURL` will be an empty string
and `grabCnx(assignedBrokerURL)` will be equivalent with `grabCnx("")`.
The code will look more clear and less code changes are needed.
--
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-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] [improve] PIP-307: Use assigned broker URL hints during broker reconnection [pulsar-client-go]
BewareMyPower commented on code in PR #1208:
URL: https://github.com/apache/pulsar-client-go/pull/1208#discussion_r1576526236
##
pulsar/producer_partition.go:
##
@@ -429,12 +453,22 @@ func (p *partitionProducer) reconnectToBroker() {
return
}
- if p.options.BackoffPolicy == nil {
+ var assignedBrokerURL string
+
+ if connectionClosed != nil && connectionClosed.HasURL() {
+ delayReconnectTime = 0
+ assignedBrokerURL = connectionClosed.assignedBrokerURL
+ connectionClosed = nil // Only attempt once
Review Comment:
How about moving the `assignedBrokerURL` out of the for loop?
```go
var assignedBrokerURL string
if connectionClosed != nil && connectionClosed.HasURL() {
assignedBrokerURL = connectionClosed.assignedBrokerURL
}
for maxRetry != 0 {
// ...
if p.options.BackoffPolicy == nil {
delayReconnectTime = defaultBackoff.Next()
} else {
delayReconnectTime = p.options.BackoffPolicy.Next()
}
```
When `connectionClosed` is nil, `assignedBrokerURL` will be an empty string
and `grabCnx(assignedBrokerURL)` will be equivalent with `grabCnx("")`.
The code will look more clear and less code changes are needed.
--
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-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] [improve] PIP-307: Use assigned broker URL hints during broker reconnection [pulsar-client-go]
BewareMyPower commented on code in PR #1208:
URL: https://github.com/apache/pulsar-client-go/pull/1208#discussion_r1576520011
##
pulsar/producer_partition.go:
##
@@ -429,12 +453,22 @@ func (p *partitionProducer) reconnectToBroker() {
return
}
- if p.options.BackoffPolicy == nil {
+ var assignedBrokerURL string
+
+ if connectionClosed != nil && connectionClosed.HasURL() {
+ delayReconnectTime = 0
+ assignedBrokerURL = connectionClosed.assignedBrokerURL
+ connectionClosed = nil // Only attempt once
Review Comment:
Oh I see. I was so blind that I missed the `for maxRetry != 0` loop before.
(BTW, the loop is too long)
--
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-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] [improve] PIP-307: Use assigned broker URL hints during broker reconnection [pulsar-client-go]
dragosvictor commented on code in PR #1208:
URL: https://github.com/apache/pulsar-client-go/pull/1208#discussion_r1576510648
##
pulsar/producer_partition.go:
##
@@ -429,12 +453,22 @@ func (p *partitionProducer) reconnectToBroker() {
return
}
- if p.options.BackoffPolicy == nil {
+ var assignedBrokerURL string
+
+ if connectionClosed != nil && connectionClosed.HasURL() {
+ delayReconnectTime = 0
+ assignedBrokerURL = connectionClosed.assignedBrokerURL
+ connectionClosed = nil // Only attempt once
Review Comment:
My understanding here is that if `grabCnx` fails, the loop handles the error
by increasing the reconnect delay and retrying up to a configurable number of
times, before giving up.
--
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-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] [improve] PIP-307: Use assigned broker URL hints during broker reconnection [pulsar-client-go]
dragosvictor commented on code in PR #1208:
URL: https://github.com/apache/pulsar-client-go/pull/1208#discussion_r1576510648
##
pulsar/producer_partition.go:
##
@@ -429,12 +453,22 @@ func (p *partitionProducer) reconnectToBroker() {
return
}
- if p.options.BackoffPolicy == nil {
+ var assignedBrokerURL string
+
+ if connectionClosed != nil && connectionClosed.HasURL() {
+ delayReconnectTime = 0
+ assignedBrokerURL = connectionClosed.assignedBrokerURL
+ connectionClosed = nil // Only attempt once
Review Comment:
My understanding here is that if `grabCnx` fails, the loop handles the error
by increasing the reconnect delay and trying up to a configurable number of
times, before giving up.
--
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-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] [improve] PIP-307: Use assigned broker URL hints during broker reconnection [pulsar-client-go]
dragosvictor commented on code in PR #1208:
URL: https://github.com/apache/pulsar-client-go/pull/1208#discussion_r1576507047
##
pulsar/producer_partition.go:
##
@@ -429,12 +453,22 @@ func (p *partitionProducer) reconnectToBroker() {
return
}
- if p.options.BackoffPolicy == nil {
+ var assignedBrokerURL string
+
+ if connectionClosed != nil && connectionClosed.HasURL() {
+ delayReconnectTime = 0
+ assignedBrokerURL = connectionClosed.assignedBrokerURL
+ connectionClosed = nil // Only attempt once
Review Comment:
There is a for loop wrapping around the reconnection attempts:
```go
for maxRetry != 0 {
if p.getProducerState() != producerReady {
// Producer is already closing
p.log.Info("producer state not ready, exit reconnect")
return
}
var assignedBrokerURL string
if connectionClosed != nil && connectionClosed.HasURL() {
delayReconnectTime = 0
assignedBrokerURL = connectionClosed.assignedBrokerURL
connectionClosed = nil // Only attempt once
} else if p.options.BackoffPolicy == nil {
delayReconnectTime = defaultBackoff.Next()
} else {
delayReconnectTime = p.options.BackoffPolicy.Next()
}
...
```
Setting the variable to nil guarantees we will only attempt connection to
the assigned broker once. If that fails, for whatever reason, we revert back to
performing topic lookups in order to locate the broker serving the topic.
--
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-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] [improve] PIP-307: Use assigned broker URL hints during broker reconnection [pulsar-client-go]
BewareMyPower commented on code in PR #1208:
URL: https://github.com/apache/pulsar-client-go/pull/1208#discussion_r1576335249
##
pulsar/producer_partition.go:
##
@@ -429,12 +453,22 @@ func (p *partitionProducer) reconnectToBroker() {
return
}
- if p.options.BackoffPolicy == nil {
+ var assignedBrokerURL string
+
+ if connectionClosed != nil && connectionClosed.HasURL() {
+ delayReconnectTime = 0
+ assignedBrokerURL = connectionClosed.assignedBrokerURL
+ connectionClosed = nil // Only attempt once
Review Comment:
Sorry I didn't get it. Could you explain it with more details?
> if the initial connection attempt fails with the given assigned URL
Did you mean `p.grabCnx(assignedBrokerURL)` failed? If so,
`reconnectToBroker` would just return and would be called when the next time
`ConnectionClosed` is called.
How can assigning a local variable (`connectionClosed`) to nil just resume
the established reconnection?
--
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-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] [improve] PIP-307: Use assigned broker URL hints during broker reconnection [pulsar-client-go]
dragosvictor commented on code in PR #1208:
URL: https://github.com/apache/pulsar-client-go/pull/1208#discussion_r1575673797
##
pulsar/producer_partition.go:
##
@@ -429,12 +453,22 @@ func (p *partitionProducer) reconnectToBroker() {
return
}
- if p.options.BackoffPolicy == nil {
+ var assignedBrokerURL string
+
+ if connectionClosed != nil && connectionClosed.HasURL() {
+ delayReconnectTime = 0
+ assignedBrokerURL = connectionClosed.assignedBrokerURL
+ connectionClosed = nil // Only attempt once
Review Comment:
Assigning the value to nil makes it resume the established reconnection
logic if the initial connection attempt fails with the given assigned URL. I
tried to capture this idea in the comment.
--
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-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] [improve] PIP-307: Use assigned broker URL hints during broker reconnection [pulsar-client-go]
BewareMyPower commented on code in PR #1208:
URL: https://github.com/apache/pulsar-client-go/pull/1208#discussion_r1575606991
##
pulsar/producer_partition.go:
##
@@ -429,12 +453,22 @@ func (p *partitionProducer) reconnectToBroker() {
return
}
- if p.options.BackoffPolicy == nil {
+ var assignedBrokerURL string
+
+ if connectionClosed != nil && connectionClosed.HasURL() {
+ delayReconnectTime = 0
+ assignedBrokerURL = connectionClosed.assignedBrokerURL
+ connectionClosed = nil // Only attempt once
Review Comment:
`connectionClosed` comes from the `ConnectionClosed` method through the
`p.connectClosedCh` channel. Assigning `nil` to this local variable seems
meaningless.
--
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-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] [improve] PIP-307: Use assigned broker URL hints during broker reconnection [pulsar-client-go]
BewareMyPower commented on code in PR #1208: URL: https://github.com/apache/pulsar-client-go/pull/1208#discussion_r1575573046 ## pulsar/internal/pulsar_proto/PulsarApi.pb.go: ## @@ -18,8 +18,8 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.28.1 -// protocv3.21.9 +// protoc-gen-go v1.31.0 +// protocv4.24.3 Review Comment: Yeah, when I regenerated it locally, I also found only the version comments were changed. -- 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-unsubscr...@pulsar.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [improve] PIP-307: Use assigned broker URL hints during broker reconnection [pulsar-client-go]
dragosvictor commented on code in PR #1208:
URL: https://github.com/apache/pulsar-client-go/pull/1208#discussion_r1575129127
##
pulsar/internal/lookup_service.go:
##
@@ -30,8 +30,9 @@ import (
// LookupResult encapsulates a struct for lookup a request, containing two
parts: LogicalAddr, PhysicalAddr.
type LookupResult struct {
- LogicalAddr *url.URL
- PhysicalAddr *url.URL
+ IsProxyThroughServiceURL bool
Review Comment:
You're right, it's never used outside of `lookupService.GetBrokerAddress`,
where it is retrieved and consumed at the same time. Removed, thanks!
--
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-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] [improve] PIP-307: Use assigned broker URL hints during broker reconnection [pulsar-client-go]
dragosvictor commented on code in PR #1208: URL: https://github.com/apache/pulsar-client-go/pull/1208#discussion_r1575070621 ## pulsar/internal/pulsar_proto/PulsarApi.pb.go: ## @@ -18,8 +18,8 @@ // Code generated by protoc-gen-go. DO NOT EDIT. // versions: -// protoc-gen-go v1.28.1 -// protocv3.21.9 +// protoc-gen-go v1.31.0 +// protocv4.24.3 Review Comment: Done! Note that it really only changed the version comments, everything else stayed the same: https://github.com/apache/pulsar-client-go/pull/1208/commits/880d732ed54d774b539d09897f273e24e5b7608f. -- 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-unsubscr...@pulsar.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
Re: [PR] [improve] PIP-307: Use assigned broker URL hints during broker reconnection [pulsar-client-go]
dragosvictor commented on code in PR #1208:
URL: https://github.com/apache/pulsar-client-go/pull/1208#discussion_r1575072107
##
pulsar/producer_partition.go:
##
@@ -373,10 +390,14 @@ func (p *partitionProducer) GetBuffer() internal.Buffer {
return b
}
-func (p *partitionProducer) ConnectionClosed() {
+func (p *partitionProducer) ConnectionClosed(closeProducer
*pb.CommandCloseProducer) {
Review Comment:
Good catch, fixed!
--
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-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] [improve] PIP-307: Use assigned broker URL hints during broker reconnection [pulsar-client-go]
BewareMyPower commented on code in PR #1208:
URL: https://github.com/apache/pulsar-client-go/pull/1208#discussion_r1574823910
##
pulsar/internal/lookup_service.go:
##
@@ -30,8 +30,9 @@ import (
// LookupResult encapsulates a struct for lookup a request, containing two
parts: LogicalAddr, PhysicalAddr.
type LookupResult struct {
- LogicalAddr *url.URL
- PhysicalAddr *url.URL
+ IsProxyThroughServiceURL bool
Review Comment:
It seems `IsProxyThroughServiceURL` is never used?
##
pulsar/producer_partition.go:
##
@@ -373,10 +390,14 @@ func (p *partitionProducer) GetBuffer() internal.Buffer {
return b
}
-func (p *partitionProducer) ConnectionClosed() {
+func (p *partitionProducer) ConnectionClosed(closeProducer
*pb.CommandCloseProducer) {
Review Comment:
I see `ConnectionClosed(nil)` could be called in `connection.go`, should you
perform the null check here?
--
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-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] [improve] PIP-307: Use assigned broker URL hints during broker reconnection [pulsar-client-go]
dragosvictor commented on code in PR #1208:
URL: https://github.com/apache/pulsar-client-go/pull/1208#discussion_r1571392492
##
pulsar/consumer_partition.go:
##
@@ -1358,10 +1358,13 @@ func createEncryptionContext(msgMeta
*pb.MessageMetadata) *EncryptionContext {
return
}
-func (pc *partitionConsumer) ConnectionClosed() {
+func (pc *partitionConsumer) ConnectionClosed(closeConsumer
*pb.CommandCloseConsumer) {
// Trigger reconnection in the consumer goroutine
pc.log.Debug("connection closed and send to connectClosedCh")
- pc.connectClosedCh <- connectionClosed{}
+ pc.connectClosedCh <- {
Review Comment:
Done!
##
pulsar/producer_partition.go:
##
@@ -221,14 +221,25 @@ func newPartitionProducer(client *client, topic string,
options *ProducerOptions
return p, nil
}
-func (p *partitionProducer) grabCnx() error {
- lr, err := p.client.lookupService.Lookup(p.topic)
+func (p *partitionProducer) lookupTopic(brokerURL, brokerURLTLS string)
(*internal.LookupResult, error) {
+ if len(brokerURL) == 0 && len(brokerURLTLS) == 0 {
Review Comment:
Done!
##
pulsar/internal/lookup_service.go:
##
@@ -284,12 +291,24 @@ type httpLookupService struct {
metrics *Metrics
}
-func (h *httpLookupService) getBrokerAddress(ld *httpLookupData)
(logicalAddress *url.URL,
+func (h *httpLookupService) GetBrokerAddress(brokerServiceURL,
brokerServiceURLTls string,
Review Comment:
Done!
--
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-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
Re: [PR] [improve] PIP-307: Use assigned broker URL hints during broker reconnection [pulsar-client-go]
heesung-sn commented on code in PR #1208:
URL: https://github.com/apache/pulsar-client-go/pull/1208#discussion_r1571120345
##
pulsar/consumer_partition.go:
##
@@ -1358,10 +1358,13 @@ func createEncryptionContext(msgMeta
*pb.MessageMetadata) *EncryptionContext {
return
}
-func (pc *partitionConsumer) ConnectionClosed() {
+func (pc *partitionConsumer) ConnectionClosed(closeConsumer
*pb.CommandCloseConsumer) {
// Trigger reconnection in the consumer goroutine
pc.log.Debug("connection closed and send to connectClosedCh")
- pc.connectClosedCh <- connectionClosed{}
+ pc.connectClosedCh <- {
Review Comment:
Can we pass one of them only by checking if tls is enabled or not here? I
think we better pass one arg to the downstream funcs to simplify. (same for
CommandCloseProducer)
##
pulsar/producer_partition.go:
##
@@ -221,14 +221,25 @@ func newPartitionProducer(client *client, topic string,
options *ProducerOptions
return p, nil
}
-func (p *partitionProducer) grabCnx() error {
- lr, err := p.client.lookupService.Lookup(p.topic)
+func (p *partitionProducer) lookupTopic(brokerURL, brokerURLTLS string)
(*internal.LookupResult, error) {
+ if len(brokerURL) == 0 && len(brokerURLTLS) == 0 {
Review Comment:
can we move this logic to `lookupService.GetBrokerAddress` to reuse? (Then
maybe we need to pass additionally pass topicName to GetBrokerAddress)
##
pulsar/internal/lookup_service.go:
##
@@ -284,12 +291,24 @@ type httpLookupService struct {
metrics *Metrics
}
-func (h *httpLookupService) getBrokerAddress(ld *httpLookupData)
(logicalAddress *url.URL,
+func (h *httpLookupService) GetBrokerAddress(brokerServiceURL,
brokerServiceURLTls string,
Review Comment:
As I commented, can we pass one of these(brokerServiceURL,
brokerServiceURLTls)?
--
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-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
