src/Protocol/CommitOffset.php
public function encode(array $payloads = []): string
    {
        if (! isset($payloads['group_id'])) {
            throw new ProtocolException('given commit offset data invalid. `group_id` is undefined.');
        }

        if (! isset($payloads['data'])) {
            throw new ProtocolException('given commit data invalid. `data` is undefined.');
        }

        if (! isset($payloads['generation_id'])) {
            $payloads['generation_id'] = -1;
        }

        if (! isset($payloads['member_id'])) {
            $payloads['member_id'] = '';
        }

        if (! isset($payloads['retention_time'])) {
            $payloads['retention_time'] = -1;
        }

        $version = $this->getApiVersion(self::OFFSET_COMMIT_REQUEST);
        $header  = $this->requestHeader('kafka-php', self::OFFSET_COMMIT_REQUEST, self::OFFSET_COMMIT_REQUEST);

        $data = self::encodeString($payloads['group_id'], self::PACK_INT16);

        if ($version === self::API_VERSION1) {
            $data .= self::pack(self::BIT_B32, (string) $payloads['generation_id']);
            $data .= self::encodeString($payloads['member_id'], self::PACK_INT16);
        }

        if ($version === self::API_VERSION2) {
            $data .= self::pack(self::BIT_B32, (string) $payloads['generation_id']);
            $data .= self::encodeString($payloads['member_id'], self::PACK_INT16);
            $data .= self::pack(self::BIT_B64, (string) $payloads['retention_time']);
        }

        $data .= self::encodeArray($payloads['data'], [$this, 'encodeTopic']);
        $data  = self::encodeString($header . $data, self::PACK_INT32);

        return $data;
    }
public function encode(array $payloads = []): string
    {
        if (! isset($payloads['group_id'])) {
            throw new ProtocolException('given commit offset data invalid. `group_id` is undefined.');
        }

        if (! isset($payloads['data'])) {
            throw new ProtocolException('given commit data invalid. `data` is undefined.');
        }

        if (! isset($payloads['generation_id'])) {
            $payloads['generation_id'] = -1;
        }

        if (! isset($payloads['member_id'])) {
            $payloads['member_id'] = '';
        }

        if (! isset($payloads['retention_time'])) {
            $payloads['retention_time'] = -1;
        }

        $version = $this->getApiVersion(self::OFFSET_COMMIT_REQUEST);
        $header  = $this->requestHeader('kafka-php', self::OFFSET_COMMIT_REQUEST, self::OFFSET_COMMIT_REQUEST);

        $data = self::encodeString($payloads['group_id'], self::PACK_INT16);

        if ($version === self::API_VERSION1) {
            $data .= self::pack(self::BIT_B32, (string) $payloads['generation_id']);
            $data .= self::encodeString($payloads['member_id'], self::PACK_INT16);
        }

        if ($version === self::API_VERSION2) {
            $data .= self::pack(self::BIT_B32, (string) $payloads['generation_id']);
            $data .= self::encodeString($payloads['member_id'], self::PACK_INT16);
            $data .= self::pack(self::BIT_B64, (string) $payloads['retention_time']);
        }

        $data .= self::encodeArray($payloads['data'], [$this, 'encodeTopic']);
        $data  = self::encodeString($header . $data, self::PACK_INT32);

        return $data;
    }
@param mixed[] $payloads

@throws ProtocolException
@throws NotSupported