Method encode

produce_request.go:72–161

(pe packetEncoder)

}

func (r *ProduceRequest) encode(pe packetEncoder) error {
	if r.Version >= 3 {
		if err := pe.putNullableString(r.TransactionalID); err != nil {
			return err
		}
	}
	pe.putInt16(int16(r.RequiredAcks))
	pe.putInt32(r.Timeout)
	metricRegistry := pe.metricRegistry()
	var batchSizeMetric metrics.Histogram
	var compressionRatioMetric metrics.Histogram
	if metricRegistry != nil {
		batchSizeMetric = getOrRegisterHistogram("batch-size", metricRegistry)
		compressionRatioMetric = getOrRegisterHistogram("compression-ratio", metricRegistry)
	}
	totalRecordCount := int64(0)

	err := pe.putArrayLength(len(r.records))
	if err != nil {
		return err
	}

	for topic, partitions := range r.records {
		err = pe.putString(topic)
		if err != nil {
			return err
		}
		err = pe.putArrayLength(len(partitions))
		if err != nil {
			return err
		}
		topicRecordCount := int64(0)
		var topicCompressionRatioMetric metrics.Histogram
		var topicBatchSizeMetric metrics.Histogram
		if metricRegistry != nil {
			topicCompressionRatioMetric = getOrRegisterTopicHistogram("compression-ratio", topic, metricRegistry)
			topicBatchSizeMetric = getOrRegisterTopicHistogram("batch-size", topic, metricRegistry)
		}
		for id, records := range partitions {
			startOffset := pe.offset()
			pe.putInt32(id)
			if r.Version >= 9 {
				// compact bytes: size the records with a prep pass to write
				// the uvarint prefix, then encode the batch non-flexibly
				var prep prepEncoder
				if err = records.encode(&prep); err != nil {
					return err
				}
				pe.putUVarint(uint64(prep.length) + 1)
				if err = records.encode(downgradeFlexibleEncoder(pe)); err != nil {
					return err
				}
			} else {
				pe.push(&lengthField{})
				err = records.encode(pe)
				if err != nil {
					return err
				}