Function decodeMemberMetadata

describegroups.go:168–224

decodeMemberMetadata converts raw metadata bytes to a DescribeGroupsResponseMemberMetadata struct. See https://github.com/apache/kafka/blob/2.4/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerProtocol.java#L49 for protocol details.

(rawMetadata []byte)

// See https://github.com/apache/kafka/blob/2.4/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerProtocol.java#L49
// for protocol details.
func decodeMemberMetadata(rawMetadata []byte) (DescribeGroupsResponseMemberMetadata, error) {
	mm := DescribeGroupsResponseMemberMetadata{}

	if len(rawMetadata) == 0 {
		return mm, nil
	}

	buf := bytes.NewBuffer(rawMetadata)
	bufReader := bufio.NewReader(buf)
	remain := len(rawMetadata)

	var err error
	var version16 int16

	if remain, err = readInt16(bufReader, remain, &version16); err != nil {
		return mm, err
	}
	mm.Version = int(version16)

	if remain, err = readStringArray(bufReader, remain, &mm.Topics); err != nil {
		return mm, err
	}
	if remain, err = readBytes(bufReader, remain, &mm.UserData); err != nil {
		return mm, err
	}

	if mm.Version == 1 && remain > 0 {
		fn := func(r *bufio.Reader, size int) (fnRemain int, fnErr error) {
			op := DescribeGroupsResponseMemberMetadataOwnedPartition{}
			if fnRemain, fnErr = readString(r, size, &op.Topic); fnErr != nil {
				return
			}

			ps := []int32{}
			if fnRemain, fnErr = readInt32Array(r, fnRemain, &ps); fnErr != nil {
				return
			}

			for _, p := range ps {
				op.Partitions = append(op.Partitions, int(p))
			}

			mm.OwnedPartitions = append(mm.OwnedPartitions, op)
			return
		}

		if remain, err = readArrayWith(bufReader, remain, fn); err != nil {
			return mm, err
		}
	}

	if remain != 0 {
		return mm, fmt.Errorf("Got non-zero number of bytes remaining: %d", remain)
	}

	return mm, nil
}