Send messages and errors directly in consumer mock #555
Conversation
The mock PartitionConsumer sent messages to an expectations channel on calls to YieldMessage and YieldError. This led to race conditions in tests if pc.Messages was called before the handleExpectations goroutine added a message to the channel. Messages and errors are now sent to their respective channels directly, and the expectations channel & goroutine are removed altogether.
| @@ -289,7 +261,13 @@ func (pc *PartitionConsumer) HighWaterMarkOffset() int64 { | |||
| // reasons forthis not to happen. ou can call ExpectMessagesDrainedOnClose so it will | |||
| // verify that the channel is empty on close. | |||
| func (pc *PartitionConsumer) YieldMessage(msg *sarama.ConsumerMessage) { | |||
| pc.expectations <- &consumerExpectation{Msg: msg} | |||
| pc.highWaterMarkOffset += 1 | |||
There was a problem hiding this comment.
Does this still need to be atomic and/or locked somehow? I guess it depends on if people expect to be able to call this method multiple times concurrently, but I don't know if that's true (it was safe to call the old one concurrently).
There was a problem hiding this comment.
It's atomic in sarama, so I prefer the mock to be atomic as well to match the sarama API as well as possible.
There was a problem hiding this comment.
I removed the atomic operations because it's no longer executed in a goroutine... then again, handleExpectations uses a for loop anyway, so that alone would not lead to race conditions. So I guess it's not totally clear to me why they were there.
If atomic operations are needed in case someone calls code concurrently, that means that plain assignments are never safe in public APIs.
There was a problem hiding this comment.
In sarama it's atomic, because it's possible to have the processing lag monitoring in a separate goroutine. So it's not so much about internal use inside the mock, but more about how the API is used.
|
One thought, but generally LGTM. |
|
👍, just need non-atomic increment needs fixing. |
|
switched back to the atomic operations. |
Send messages and errors directly in consumer mock
|
Thanks! |
| @@ -289,7 +260,13 @@ func (pc *PartitionConsumer) HighWaterMarkOffset() int64 { | |||
| // reasons forthis not to happen. ou can call ExpectMessagesDrainedOnClose so it will | |||
| // verify that the channel is empty on close. | |||
| func (pc *PartitionConsumer) YieldMessage(msg *sarama.ConsumerMessage) { | |||
| pc.expectations <- &consumerExpectation{Msg: msg} | |||
| atomic.AddInt64(&pc.highWaterMarkOffset, 1) | |||
There was a problem hiding this comment.
two concurrent calls to this method can still interleave, resulting in incorrect msg.Offset values
There was a problem hiding this comment.
Are you suggesting that lock/unlock should be added here too?
There was a problem hiding this comment.
I think it is necessary? atomic.AddInt64 returns the new value, so you could collapse the atomic calls and reduce the window, but it would still be possible to interleave method calls such that messages ended up in the channel with the wrong order (e.g. offsets 1 2 3 4 instead of 1 2 3 4). A lock seems the obvious way to force the highWaterMarkOffset change to occur synchronously with the channel write.
Note that the variable will still need to be edited atomically in order to avoid locking HighWaterMarkOffset() too.
There was a problem hiding this comment.
I could add it in, this is already merged... will it still work on this branch, or should I do another?
There was a problem hiding this comment.
another branch is probably simpler
The mock
PartitionConsumersent messages to an expectations channelon calls to
YieldMessageandYieldError. This led to race conditionsin tests if
pc.Messageswas called before thehandleExpectationsgoroutine added a message to the channel.
This PR sends message and errors to their respective channels
directly. The expectations channel & goroutine are removed
altogether, simplifying the code.
This resolves issue #553.
@eapache @wvanbergen