Auto update the client to handle changes in number of partitions #221
Conversation
| } | ||
| consumer.ticker = time.NewTicker(duration) | ||
|
|
||
| go func() { |
There was a problem hiding this comment.
Should we only start this if the topic is partitioned?
There was a problem hiding this comment.
Problem is that the TopicPartitions() will just return a list, so we don't necessarely know if it's 1 partition or no partitions. In any case, if the topic is not partitioned, it would not be possible to create a partitioned topic with same name.
| p.producers = make([]Producer, numPartitions) | ||
| p.ticker = time.NewTicker(partitionsAutoDiscoveryInterval) | ||
|
|
||
| go func() { |
There was a problem hiding this comment.
Will this run for non-partitioned topics as well? Is that ok?
| return p, nil | ||
| } | ||
|
|
||
| func (p *producer) internalCreatePartitionsProducers() error { |
There was a problem hiding this comment.
Same comment as consumers
|
General question about how locks are handled in Go, when multiple threads are waiting on a lock. In what order are the locks released? FIFO? Or just everyone attempts to acquire and first one wins? I am asking because I am wondering if there are going to be any thread starvation going on for the thread that periodically checks for new partitions especially when "sendAsync" can be called quite frequently. |
|
Yes, I was trying to avoid getting the extra mutex on the write path, though I think it shouldn't be too bad.
I haven't really seen it specified, though a mutex always should include some sort of queuing for fairness. I'm not worried about that aspect. Also, the thread that updates the list of producers is not "critical" in the sense that it only needs to get executed eventually, with no hard time bound. |
Motivation
Client should automatically discover new partitions as they're added.