Method encode

offset_fetch_request.go:72–165

(pe packetEncoder)

}

func (r *OffsetFetchRequest) encode(pe packetEncoder) (err error) {
	if r.Version < 0 || r.Version > 8 {
		return PacketEncodingError{"invalid or unsupported OffsetFetchRequest version field"}
	}

	if r.RequireStable && r.Version < 7 {
		return PacketEncodingError{"requireStable is not supported. use version 7 or later"}
	}

	if r.Version >= 8 {
		if len(r.Groups) == 0 {
			return PacketEncodingError{"version 8 or later requires Groups to be populated"}
		}
		if err := pe.putArrayLength(len(r.Groups)); err != nil {
			return err
		}
		for _, g := range r.Groups {
			if err := pe.putString(g.GroupId); err != nil {
				return err
			}

			// nil Partitions encodes as null topics array (fetch all)
			topicCount := len(g.Partitions)
			if g.Partitions == nil {
				topicCount = -1
			}
			if err := pe.putArrayLength(topicCount); err != nil {
				return err
			}
			for topic, partitions := range g.Partitions {
				if err := pe.putString(topic); err != nil {
					return err
				}
				if err := pe.putInt32Array(partitions); err != nil {
					return err
				}
				pe.putEmptyTaggedFieldArray()
			}

			pe.putEmptyTaggedFieldArray()
		}

		pe.putBool(r.RequireStable)
		pe.putEmptyTaggedFieldArray()
		return nil
	}

	if len(r.Groups) > 1 {
		return PacketEncodingError{"multiple groups require version 8 or later"}
	}

	consumerGroup := r.ConsumerGroup
	partitions := r.partitions
	if consumerGroup == "" && len(r.Groups) == 1 {
		consumerGroup = r.Groups[0].GroupId
		partitions = r.Groups[0].Partitions
	}