Fix race condition on OffsetManager shutdown

[eapache]

If updateCommitted ran between the Unlock call and the read from clean
then we might miss the transition message and hang. Instead, read
unconditionally from clean and launch a second goroutine that writes a
message to it if we were already clean.

The additional local cancel is necessary to avoid leaking the second (writing)
goroutine in the event that the main shutdown goroutine picks up a real
transitional event before our extra writer even runs; without it the writer
could block writing forever.

@wvanbergen @hasnickl fixes #656

[hasnickl]

If the first go func runs while pom is dirty, updateCommitted could run and finish before the second go func receives on clean

A possible solution is to have an internal flag like pom.shutdown. AsyncClose can lock pom, set shutdown to true, start the go routine to read from clean and unlock.

updateCommitted could then change the select with default to conditionally start a go routine:

if pom.shutdown {
  go func(){
    pom.clean <- none{}
  }
}

perhaps MarkOffset would also check this flag to prevent state going to dirty again while shutting down.

Any thoughts on this?

[eapache]

If the first go func runs while pom is dirty, updateCommitted could run and finish before the second go func receives on clean

You're right, I must have been asleep when I cooked up this solution.