watermill
watermill copied to clipboard
[watermill-kafka] `Subscriber.Close()` seems not concurrent safe
kafka.Subscriber.Close()
seems not concurrent safe.
close(s.closing)
may be executed multiple times when called concurrently,
causing panic: close of closed channel
.
https://github.com/ThreeDotsLabs/watermill-kafka/blob/9242094c8fd755f501295c65515f2d47bb122962/pkg/kafka/subscriber.go#L468
func (s *Subscriber) Close() error {
if s.closed {
return nil
}
s.closed = true
close(s.closing)
see alsao:
- https://github.com/ThreeDotsLabs/watermill-amqp/blob/f8efc8147ada84ac953420f042f87995168a829a/pkg/amqp/subscriber.go#L42-L45
- https://github.com/ThreeDotsLabs/watermill-amqp/blob/f8efc8147ada84ac953420f042f87995168a829a/pkg/amqp/subscriber.go#L78-L81
- https://github.com/ThreeDotsLabs/watermill-nats/blob/1a6d735bb37b3f97ab822abacc233428a56041e6/pkg/nats/subscriber.go#L385-L386