Method ProduceSync

pkg/ingest/writer_client.go:270–337

ProduceSync produces records to Kafka and returns once all records have been successfully committed, or an error occurred. This function honors the configure max buffered bytes and refuse to produce a record, returnin kgo.ErrMaxBuffered, if the configured limit is reached.

(ctx context.Context, records []*kgo.Record)

// This function honors the configure max buffered bytes and refuse to produce a record, returnin kgo.ErrMaxBuffered,
// if the configured limit is reached.
func (c *Producer) ProduceSync(ctx context.Context, records []*kgo.Record) kgo.ProduceResults {
	var (
		remaining = atomic.NewInt64(int64(len(records)))
		done      = make(chan struct{})
		resMx     sync.Mutex
		res       = make(kgo.ProduceResults, 0, len(records))
	)

	// As a safety mechanism, we want to make sure that the context is not already canceled or its deadline exceeded.
	// The reason is that once we buffer records to the Kafka client (later in this function), these records will be
	// sent to the Kafka backend regardless the context is canceled or not. There's no way, in the Kafka client, to
	// pull out records from a batch buffer. So, if the context is canceled, we circuit break instead of buffering
	// records that may be sent to the Kafka backend, but that the caller will not know about because context is done.
	if err := ctx.Err(); err != nil {
		recordsCount := float64(len(records))
		c.produceRecordsFailuresTotal.WithLabelValues("cancelled-before-producing").Add(recordsCount)
		// We wrap the error to make it cristal clear where the context canceled/timeout comes from.
		return kgo.ProduceResults{{Err: fmt.Errorf("skipped producing Kafka records because context is already done: %w", err)}}
	}

	onProduceDone := func(r *kgo.Record, err error) {
		if c.maxBufferedBytes > 0 {
			c.bufferedBytes.Add(-int64(len(r.Value)))
		}

		resMx.Lock()
		res = append(res, kgo.ProduceResult{Record: r, Err: err})
		resMx.Unlock()

		if err != nil {
			c.produceRecordsFailuresTotal.WithLabelValues(produceErrReason(err)).Inc()
		}

		// In case of error we'll wait for all responses anyway before returning from produceSync().
		// It allows us to keep code easier, given we don't expect this function to be frequently
		// called with multiple records.
		if remaining.Dec() == 0 {
			close(done)
		}
	}

	for _, record := range records {
		// Fast fail if the Kafka client buffer is full. Buffered bytes counter is decreased onProducerDone().
		if c.maxBufferedBytes > 0 && c.bufferedBytes.Add(int64(len(record.Value))) > c.maxBufferedBytes {
			onProduceDone(record, kgo.ErrMaxBuffered)
			continue
		}

		// We use a new context to avoid that other Produce() may be cancelled when this call's context is
		// canceled. It's important to note that cancelling the context passed to Produce() doesn't actually
		// prevent the data to be sent over the wire (because it's never removed from the buffer) but in some
		// cases may cause all requests to fail with context cancelled.
		//
		// Produce() may theoretically block if the buffer is full, but we configure the Kafka client with
		// unlimited buffer because we implement the buffer limit ourselves (see maxBufferedBytes). This means
		// Produce() should never block for us in practice.
		c.Produce(context.WithoutCancel(ctx), record, onProduceDone)
	}