OffsetManager.Close() race condition

[hasnickl]

When closing OffsetManager, it can hang waiting for state to become "clean".

The Close method uses AsyncClose internally. At L360, AsyncClose waits for a message on the clean chan before closing the dying chan:

        pom.lock.Lock()
        dirty := pom.dirty
        pom.lock.Unlock()

        if dirty {
            <-pom.clean
        }

        close(pom.dying)

However, updateCommitted can execute between AsyncClose()'s pom.lock.Unlock() and receive on the clean chan:

    pom.lock.Lock()
    defer pom.lock.Unlock()
...
        select {
        case pom.clean <- none{}:
        default:
        }

The Close() method drains pom.errors at L371. This channel does not close until pom.dying is closed, which is selected in the mainLoop at L215