code: func (p *Producer) loop() {
  	buf := make([]*k.PutRecordsRequestEntry, 0, p.BufferSize)
  	tick := time.NewTicker(p.FlushInterval)
  	drain := false
  
  	defer tick.Stop()
  	defer close(p.done)
  
  	for {
  		select {
  		case record := <-p.records:
  			buf = append(buf, record)
  
  			if len(buf) >= p.BufferSize {
  				p.flush(buf, "buffer size")
  				buf = nil
  			}
  
  			if drain && len(p.records) == 0 {
  				p.Logger.Info("drained")
  				return
  			}
  		case <-tick.C:
  			if len(buf) > 0 {
  				p.flush(buf, "interval")
  				buf = nil
  			}
  		case <-p.done:
  			drain = true
  			if len(buf) > 0 {
  				p.flush(buf, "stopping")
  				buf = nil
  			}
  
  			if len(p.records) == 0 {
  				return
  			}
  		}
  	}
  }
docs: // loop and flush at the configured interval, or when the buffer is exceeded.