Log errors after failing to send acks #3855
Conversation
This will help in debugging issues when the coordinator fails to send an ack to the aggregator.
src/msg/consumer/consumer.go
Outdated
| } | ||
| c.w.Flush() | ||
| c.Unlock() | ||
| } | ||
|
|
||
| func (c *consumer) encodeAckWithLock(ackLen int) error { | ||
| func (c *consumer) sendAckWithLock(ackLen int) error { |
There was a problem hiding this comment.
seemed like a more appropriate method name, since it actually sends the ack over the network.
| c.ackPb.Metadata = c.ackPb.Metadata[:0] | ||
| if err != nil { | ||
| c.m.ackEncodeError.Inc(1) | ||
| return err | ||
| log.Error("failed to encode ack. client will retry sending message.", zap.Error(err)) |
There was a problem hiding this comment.
previously we would close the connection if we failed to encode, which doesn't make sense since it has nothing to do with the conn.
Codecov Report
@@ Coverage Diff @@
## master #3855 +/- ##
======================================
Coverage 56.9% 56.9%
======================================
Files 555 555
Lines 63390 63390
======================================
Hits 36079 36079
Misses 24121 24121
Partials 3190 3190
Flags with carried forward coverage won't be shown. Click here to find out more. Continue to review full report at Codecov.
|
* master: [agg] Use timestamp (not start aligned) for expiring forward versions (#3922) [tests] Add support for calls to label APIs in resources.Coordinator (#3916) [tests] Convert repair_and_replication Docker Integration Test to In-process (#3903) Always Close the conn if failed to write acks (#3855) [m3msg] Add receive and handle latency to consumers (#3920)
This will help in debugging issues when the coordinator fails to send an
ack to the aggregator.
What this PR does / why we need it:
Fixes #
Special notes for your reviewer:
Does this PR introduce a user-facing and/or backwards incompatible change?:
Does this PR require updating code package or user-facing documentation?: