PavelZeger commented on code in PR #1490:
URL: https://github.com/apache/pulsar-client-go/pull/1490#discussion_r3247253064
##########
pulsar/consumer.go:
##########
@@ -222,6 +222,28 @@ type ConsumerOptions struct {
// MaxReconnectToBroker sets the maximum retry number of
reconnectToBroker. (default: ultimate)
MaxReconnectToBroker *uint
+ // MaxReconnectToBrokerListener is called when the consumer gives up on
reconnecting to the
+ // broker. The consumer argument is the parent consumer, and err is the
last connection error.
+ // Use this callback to detect silent failure and take recovery action
(e.g. recreate the
+ // consumer). The callback fires at most once per reconnect cycle, in
either of two cases:
+ // 1. The retry budget set by MaxReconnectToBroker is exhausted.
+ // 2. The broker reports a non-retriable error (e.g.
AuthorizationError, TopicNotFound,
+ // TopicTerminated, IncompatibleSchema). In this case the listener
fires regardless of
+ // whether MaxReconnectToBroker was set, since retrying cannot
recover.
+ // This callback is invoked from the partition consumer event loop, so
applications must not
+ // call consumer.Close() synchronously from within the callback — doing
so can deadlock. If
+ // closing is required, do it asynchronously (for example, in another
goroutine), or enable
+ // CloseConsumerOnMaxReconnectToBroker to let the client close the
consumer safely after the
+ // callback returns.
+ MaxReconnectToBrokerListener func(consumer Consumer, err error)
Review Comment:
Before I push the changes, I want to confirm a couple of details so I
implement it the way you have in mind.
1. Signature - should it be `OnConsumerClose(consumer Consumer)` or
`OnConsumerClose(consumer Consumer, cause error)`? The original issue #1481
asked for a way to know why the consumer gave up reconnecting, so passing the
last error as cause would keep that information. For a normal user-initiated
Close(), cause would be nil. Do you prefer this, or the simpler version without
the error?
2. When to call it - I plan to fire `OnConsumerClose` once per consumer, at
the start of `consumer.Close()` (before teardown), guarded by a `sync.Once`.
That way it fires exactly once whether the close comes from the user, from
max-reconnect exhaustion, or from a non-retriable error. Does that match what
you had in mind?
3. Breaking change - adding a method to ConsumerInterceptor will break
existing implementations. I'll update the in-repo ones and add a note to the
changelog. Are you OK with this, or would you prefer a separate optional
interface (e.g., `ConsumerCloseInterceptor`) to keep backward compatibility?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
