Sync recover #11
Sync recover #11
Conversation
| } | ||
|
|
||
| go func() { | ||
| ticker := time.NewTicker(delayProxyInterval) |
There was a problem hiding this comment.
for me it's too much nesting here. You could at least check the iterating over the waiters to
func (p *delayProxy) waitersDone() bool {
for _, r := range p.wait {
if !r() {
return false
}
}
return true
}
Then the done can be removed and the check is easier
There was a problem hiding this comment.
Actually I'd even extract the function doing the loop over the ticker altogether. It's not a closure, so there's no need to define it inline. Then just call the extracted function in a goroutine and it's much clearer
proxy.go
Outdated
| } | ||
| } | ||
| if done { | ||
| p.consumer.AddGroupPartition(p.partition) |
There was a problem hiding this comment.
missing return here? Otherwise the group partition is added to the consumer on every tick
| @@ -379,7 +379,11 @@ func (g *Processor) run() { | |||
| } | |||
|
|
|||
| case *kafka.NOP: | |||
| _ = g.pushToPartition(ev.Partition, ev) | |||
| if g.graph.joint(ev.Topic) { | |||
| _ = g.pushToPartitionView(ev.Topic, ev.Partition, ev) | |||
There was a problem hiding this comment.
sure we want to ignore the error?
There was a problem hiding this comment.
NOP is only used for testing
There was a problem hiding this comment.
can only occur in unit-test. so it's fine.
waits untils views and partitioned views are ready before processing events of the partitions