Add send error logic for connection #566
Conversation
Signed-off-by: xiaolongran <xiaolongran@tencent.com>
Signed-off-by: xiaolongran <xiaolongran@tencent.com>
Signed-off-by: xiaolongran <xiaolongran@tencent.com>
Signed-off-by: xiaolongran <xiaolongran@tencent.com>
Signed-off-by: xiaolongran <xiaolongran@tencent.com>
Signed-off-by: xiaolongran <xiaolongran@tencent.com>
|
ping @zymap PTAL |
pulsar/internal/connection.go
Outdated
| if ok { | ||
| producer.ConnectionClosed() | ||
| } else { | ||
| c.log. |
There was a problem hiding this comment.
No, this is automatically formatted by golanglint, probably for aesthetic reasons
pulsar/internal/connection.go
Outdated
| c.listenersLock.RUnlock() | ||
|
|
||
| if ok { | ||
| producer.ConnectionClosed() |
There was a problem hiding this comment.
In the Java implementation, we change the state to terminate and send the error to all pending messages, does the producer connection closed will do the same thing with that?
There was a problem hiding this comment.
Why is this called and the connection is not closed? What should the behavior be when this error happens? What do the other clients do? Calling producer.ConnectionClosed() when we don't close connection make the code confusing in my opinion.
There was a problem hiding this comment.
Nothing here is to send a connection closed signal to runEventsLoop, try to trigger the logic of reconnection
There was a problem hiding this comment.
The java client closes the connection in the default case.
Should we do the same. What should the expected client behavior be here?
pulsar/internal/connection.go
Outdated
| producerID := sendError.GetProducerId() | ||
|
|
||
| SendError: | ||
| switch *sendError.Error { |
There was a problem hiding this comment.
We also handle the ChecksumError in the java implementation, do we need to implement that?
pulsar/internal/connection.go
Outdated
| c.listenersLock.RUnlock() | ||
|
|
||
| if ok { | ||
| producer.ConnectionClosed() |
There was a problem hiding this comment.
Why is this called and the connection is not closed? What should the behavior be when this error happens? What do the other clients do? Calling producer.ConnectionClosed() when we don't close connection make the code confusing in my opinion.
pulsar/internal/connection.go
Outdated
|
|
||
| errMsg := fmt.Sprintf("server error: %s: %s", cmdError.GetError(), cmdError.GetMessage()) | ||
| request.callback(nil, errors.New(errMsg)) | ||
| break SendError |
There was a problem hiding this comment.
I'm not following why there needs to be a break to SendError?
| Warn("[HandleSendError] connection closed") | ||
| } | ||
| break SendError | ||
| default: |
There was a problem hiding this comment.
Is the pending request callback not needed in the default case?
Signed-off-by: xiaolongran <xiaolongran@tencent.com>
|
There is currently no good idea to deal with TopicTerminatedError, so it is temporarily not supported. |
| @@ -33,7 +33,9 @@ func (t *ProducerInterceptor) BeforeSend(producer pulsar.Producer, message *puls | |||
| buildAndInjectSpan(message, producer).Finish() | |||
| } | |||
|
|
|||
| func (t *ProducerInterceptor) OnSendAcknowledgement(producer pulsar.Producer, message *pulsar.ProducerMessage, msgID pulsar.MessageID) { | |||
| func (t *ProducerInterceptor) OnSendAcknowledgement(producer pulsar.Producer, | |||
There was a problem hiding this comment.
Can we revert these formatting changes since they have nothing to do with this change?
There was a problem hiding this comment.
This will cause Action CI to fail:
Error: pulsar/internal/pulsartracing/producer_interceptor.go:36: line is 136 characters (lll)
func (t *ProducerInterceptor) OnSendAcknowledgement(producer pulsar.Producer, message *pulsar.ProducerMessage, msgID pulsar.MessageID) {
Error: pulsar/internal/pulsartracing/producer_interceptor_test.go:58: line is 132 characters (lll)
func (p *mockProducer) SendAsync(context.Context, *pulsar.ProducerMessage, func(pulsar.MessageID, *pulsar.ProducerMessage, error)) {
Error: Process completed with exit code 1.
pulsar/internal/connection.go
Outdated
| c.listenersLock.RUnlock() | ||
|
|
||
| if ok { | ||
| producer.ConnectionClosed() |
There was a problem hiding this comment.
The java client closes the connection in the default case.
Should we do the same. What should the expected client behavior be here?
pulsar/internal/connection.go
Outdated
|
|
||
| switch *sendError.Error { | ||
| case pb.ServerError_NotAllowedError: | ||
| c.pendingLock.Lock() |
There was a problem hiding this comment.
This block of code is used in a couple of places now maybe we should add a function? Something like this?
func deletePendingRequest(requestID) (request, bool) {
c.pendingLock.Lock()
defer c.pendingLock.Unlock()
request, ok := c.pendingReqs[requestID]
if ok {
delete(c.pendingReqs, requestID)
}
return request, ok
}
There was a problem hiding this comment.
The java client closes the connection in the default case.
Should we do the same. What should the expected client behavior be here?
About this, Look at the comments like reconnect, look at the code like close the current connection:
// By default, for transient error, let the **reconnection** logic
// to take place and re-establish the produce again
ctx.close();
Signed-off-by: xiaolongran <xiaolongran@tencent.com>
Signed-off-by: xiaolongran <xiaolongran@tencent.com>
Signed-off-by: xiaolongran xiaolongran@tencent.com
Fixes #564
Motivation
Add command of SendError logic for connection