We should only consider the connection with rdyCount ==0. For the active connection, RDY should not be updated again.

[neezhe]

how about this ?

change

for len(possibleConns) > 0 && availableMaxInFlight > 0 {
        availableMaxInFlight--
        r.rngMtx.Lock()
        i := r.rng.Int() % len(possibleConns)
        r.rngMtx.Unlock()
        c := possibleConns[i]
        // delete
        possibleConns = append(possibleConns[:i], possibleConns[i+1:]...)
        r.log(LogLevelDebug, "(%s) redistributing RDY", c.String())
        r.updateRDY(c, 1) 
}

into

for len(possibleConns) > 0 && availableMaxInFlight > 0 {
        availableMaxInFlight--
        r.rngMtx.Lock()
        i := r.rng.Int() % len(possibleConns)
        r.rngMtx.Unlock()
        c := possibleConns[i]
        // delete
        possibleConns = append(possibleConns[:i], possibleConns[i+1:]...)
        r.log(LogLevelDebug, "(%s) redistributing RDY", c.String())
        if c.rdyCount > 0 {
            continue
        }
        r.updateRDY(c, 1) 
}

[ploxiln]

If you would like to propose a change, it is much easier to read it in context if you open a pull request.

This logic is the trickiest in nsq consumer libraries, so I do not know, without digging in quite a bit, whether this is appropriate or not. So it will be a while until you get a proper reply.

[mreiferson]

The change looks like it is specifically looking to "optimize" for a case where a connection already has a non-zero rdyCount. But for this loop, that doesn't make sense, if anything you'd want a connection where rdyCount == 0.

Thanks for the suggestion, but I'm going to close this.