Producer deadlock: overlimit=0

[guyanb]

file:produce_message.go

func (msg *produceMessage) enqueue(p *Producer) error {
    ......
    errs := make(chan error, 1)
    bp.flushRequest(p, prb, func(err error) {
        errs <- err
    })
    //:3 block
    return <-errs
}

file:producer.go

func (bp *brokerProducer) flushRequest(p *Producer, prb produceRequestBuilder, errorCb func(error)) (shutdownRequired bool) {
    ......
    switch err {
    case nil:
        break
    case EncodingError:
        ......
    default:
        overlimit := 0
        prb.reverseEach(func(msg *produceMessage) {
            //:1 if msg.reenqueue(p) return a nil, then overlimit=0
            if err := msg.reenqueue(p); err != nil {
                overlimit++
            }
        })
        //:2 if overlimit=0, not call errorCb
        if overlimit > 0 {
            errorCb(DroppedMessagesError{overlimit, nil})
        }
        return true
    }
    ......
}

func (msg *produceMessage) reenqueue(p *Producer) error {
    if !msg.retried {
        msg.retried = true
        return msg.enqueue(p)
    }
    return nil
}

[eapache]

I believe the correct fix for this is:

• have reenqueue return non-nil when msg.retried so we know it hasn't been delivered
• have flushRequest call errorCb(nil) when overlimit == 0

On inspection there are two other issues:

• duplicate of this code around line 378 of producer.go, as above
• if the response is well formed but empty (no blocks, or blocks but no partitions) then we return without ever calling errorCb - to fix this we should probably set a flag when we see a block-partition, and call errorCb if the flag is unset after the main loop

[eapache]

@burke @fw42 @wvanbergen does this make sense to you?

[wvanbergen]

I am not quite sure what the initial problem was here? :/

[fw42]

Yeah ditto, no clue what this is about :-) Can you elaborate?

[eapache]

It's a deadlock, read the inserted comments in the code block in order (1, 2, 3). #90 also includes a stack trace which I believe is the same issue.

[eapache]

Basically: enqueue expects that flushRequest calls errorCb exactly once, so that there is something on the errs channel for it to return. There are some cases where we never call errorCb (or call errorCb multiple times) which both cause a deadlock reading or writing to the errs channel.

[wvanbergen]

OK, now it makes sense to me. Your solutions makes sense to me. I wonder whether we can build tests around this.

[eapache]

https://groups.google.com/forum/#!topic/sarama-users/8smSO57Ubzg found a way to consistently reproduce the hang, but it requires some dns hackery - I will use that to at least test a local copy of the fix