a little help? about sarama

[schleppy]

I am in the exact same boat, and while I don't have a definitive answer, this is what I do know.

The error returned from QueueMessage will only be non-nil for errors in constructing the message (since it isn't actually sent at this time). SendMessage (being synchronous) will return an error more likely tied to sending / acking. The channel returned from Producer.Errors() is a buffered channel (size 16) that will often times contain nil errors. I have not determined the source for the nils being written to the channel, but they are definitely there.

From what I can tell, there is no means currently to recover the message (or batch of messages) that generated a non-nil error on the errors channel. I have started playing with possibilities for handling this - though no significant progress has been made. I hope to determine a solution as well as the messages being persisted to the Kafka queue in my case can not be lossy.

[davidbirdsong]

I spent some time reworking how data arrives at my producer so that I can apply back pressure against it. So now I'm looking at whether to just change over to SendMessage or look more closely at how QueueMessage is constructed and possibly reimplement what's going on under the covers to get messagesets working.

I might have focused on the wrong source since the errors I'm seeing are seen by others. It could be that I've mis-diagnosed my problem and just need to look into whether using snappy is the source of data loss.

[schleppy]

I spent some time last night on this issue as well. I am in the position where every message is equally important and I can't lose any of them.

What I did (testing for now) is add an additional channel, failed chan []byte to the Producer and then anywhere there is a non-nil error returned I call

func (p *Producer) sendFailed(prb produceRequestBuilder) {
    for _, msg := range prb {
        p.failed <- *msg.value
    }
}

Then I simply have a goroutine that consumes off of this channel and replays them onto the channel I am originally getting the messages from.

[davidbirdsong]

@schleppy is this a workaround to allow continued use of QueueMessage?

[schleppy]

@davidbirdsong This is a workaround to allow for handling of failed flushes.

[davidbirdsong]

Sure, but are you using Producer.SendMessage or Produser.QueueMessage?

[schleppy]

QueueMessage

[eapache]

@schleppy that's a good idea, do you want to clean it up (potentially make it configurable?) and submit it in a pull request?

@davidbirdsong are there any other issues you have (besides ones filed separately like #71) or should this be closed?

[davidbirdsong]

@eapache thanks, closing.

[schleppy]

@eapache Yeah, I will look at getting that in a pull request today or tomorrow. It currently isn't configurable, so that may take some thinking.