Method awaitBatch

writer.go:1075–1099

awaitBatch waits for a batch to either fill up or time out. If the batch is full it only stops the timer, if the timer expires it will queue the batch for writing if needed.

(batch *writeBatch)

// If the batch is full it only stops the timer, if the timer
// expires it will queue the batch for writing if needed.
func (ptw *partitionWriter) awaitBatch(batch *writeBatch) {
	select {
	case <-batch.timer.C:
		ptw.mutex.Lock()
		// detach the batch from the writer if we're still attached
		// and queue for writing.
		// Only the current batch can expire, all previous batches were already written to the queue.
		// If writeMesseages locks pw.mutex after the timer fires but before this goroutine
		// can lock pw.mutex it will either have filled the batch and enqueued it which will mean
		// pw.currBatch != batch so we just move on.
		// Otherwise, we detach the batch from the ptWriter and enqueue it for writing.
		if ptw.currBatch == batch {
			ptw.queue.Put(batch)
			ptw.currBatch = nil
		}
		ptw.mutex.Unlock()
	case <-batch.ready:
		// The batch became full, it was removed from the ptwriter and its
		// ready channel was closed. We need to close the timer to avoid
		// having it leak until it expires.
		batch.timer.Stop()
	}
	stats := ptw.w.stats()
	stats.batchQueueTime.observe(int64(time.Since(batch.time)))
}

func (ptw *partitionWriter) writeBatch(batch *writeBatch) {
	stats := ptw.w.stats()