Method Close

writer.go:555–580

Close flushes pending writes, and waits for all writes to complete before returning. Calling Close also prevents new writes from being submitted to the writer, further calls to WriteMessages and the like will fail with io.ErrClosedPipe.

()

// the writer, further calls to WriteMessages and the like will fail with
// io.ErrClosedPipe.
func (w *Writer) Close() error {
	w.mutex.Lock()
	// Marking the writer as closed here causes future calls to WriteMessages to
	// fail with io.ErrClosedPipe. Mutation of this field is synchronized on the
	// writer's mutex to ensure that no more increments of the wait group are
	// performed afterwards (which could otherwise race with the Wait below).
	w.closed = true

	// close all writers to trigger any pending batches
	for _, writer := range w.writers {
		writer.close()
	}

	for partition := range w.writers {
		delete(w.writers, partition)
	}

	w.mutex.Unlock()
	w.group.Wait()

	if w.transport != nil {
		w.transport.CloseIdleConnections()
	}

	return nil
}

// WriteMessages writes a batch of messages to the kafka topic configured on this
// writer.