Method retryHandler

async_producer.go:1574–1631

singleton effectively a "bridge" between the flushers and the dispatcher in order to avoid deadlock based on https://godoc.org/github.com/eapache/channels#InfiniteChannel

()

// effectively a "bridge" between the flushers and the dispatcher in order to avoid deadlock
// based on https://godoc.org/github.com/eapache/channels#InfiniteChannel
func (p *asyncProducer) retryHandler() {
	maxBufferLength := p.conf.Producer.Retry.MaxBufferLength
	if 0 < maxBufferLength && maxBufferLength < minFunctionalRetryBufferLength {
		maxBufferLength = minFunctionalRetryBufferLength
	}

	maxBufferBytes := p.conf.Producer.Retry.MaxBufferBytes
	if 0 < maxBufferBytes && maxBufferBytes < minFunctionalRetryBufferBytes {
		maxBufferBytes = minFunctionalRetryBufferBytes
	}

	version := 1
	if p.conf.Version.IsAtLeast(V0_11_0_0) {
		version = 2
	}

	var currentByteSize int64
	var msg *ProducerMessage
	var buf queue.Queue[*ProducerMessage]

	for {
		if buf.Length() == 0 {
			msg = <-p.retries
		} else {
			select {
			case msg = <-p.retries:
			case p.input <- buf.Peek():
				msgToRemove := buf.Remove()
				currentByteSize -= int64(msgToRemove.ByteSize(version))
				continue
			}
		}

		if msg == nil {
			return
		}

		buf.Add(msg)
		currentByteSize += int64(msg.ByteSize(version))

		if (maxBufferLength <= 0 || buf.Length() < maxBufferLength) && (maxBufferBytes <= 0 || currentByteSize < maxBufferBytes) {
			continue
		}

		msgToHandle := buf.Peek()
		if msgToHandle.flags == 0 {
			select {
			case p.input <- msgToHandle:
				buf.Remove()
				currentByteSize -= int64(msgToHandle.ByteSize(version))
			default:
				buf.Remove()
				currentByteSize -= int64(msgToHandle.ByteSize(version))
				p.returnError(msgToHandle, ErrProducerRetryBufferOverflow)
			}
		}
	}
}