some goroutine never stop, more and more goroutines

[childe]

In simple_consumer, fetch goroutine and decode goroutine maybe blocked if consume goroutine returns.