src/Protocol/Fetch.php
protected function decodeMessage(string $data, int $messageSize): ?array
    {
        if ($messageSize === 0 || strlen($data) < $messageSize) {
            return null;
        }

        $offset  = 0;
        $crc     = self::unpack(self::BIT_B32, substr($data, $offset, 4));
        $offset += 4;

        $magic = self::unpack(self::BIT_B8, substr($data, $offset, 1));
        ++$offset;

        $attr = self::unpack(self::BIT_B8, substr($data, $offset, 1));
        ++$offset;

        $timestamp  = 0;
        $backOffset = $offset;

        try { // try unpack message format v1, falling back to v0 if it fails
            if ($magic >= self::MESSAGE_MAGIC_VERSION1) {
                $timestamp = self::unpack(self::BIT_B64, substr($data, $offset, 8));
                $offset   += 8;
            }

            $keyRet  = $this->decodeString(substr($data, $offset), self::BIT_B32);
            $offset += $keyRet['length'];

            $valueRet = $this->decodeString((string) substr($data, $offset), self::BIT_B32, $attr & Produce::COMPRESSION_CODEC_MASK);
            $offset  += $valueRet['length'];

            if ($offset !== $messageSize) {
                throw new Exception(
                    'pack message fail, message len:' . $messageSize . ' , data unpack offset :' . $offset
                );
            }
        } catch (Exception $e) { // try unpack message format v0
            $offset    = $backOffset;
            $timestamp = 0;
            $keyRet    = $this->decodeString(substr($data, $offset), self::BIT_B32);
            $offset   += $keyRet['length'];

            $valueRet = $this->decodeString(substr($data, $offset), self::BIT_B32);
            $offset  += $valueRet['length'];
        }

        return [
            'length' => $offset,
            'data'   => [
                'crc'       => $crc,
                'magic'     => $magic,
                'attr'      => $attr,
                'timestamp' => $timestamp,
                'key'       => $keyRet['data'],
                'value'     => $valueRet['data'],
            ],
        ];
    }
protected function decodeMessage(string $data, int $messageSize): ?array
    {
        if ($messageSize === 0 || strlen($data) < $messageSize) {
            return null;
        }

        $offset  = 0;
        $crc     = self::unpack(self::BIT_B32, substr($data, $offset, 4));
        $offset += 4;

        $magic = self::unpack(self::BIT_B8, substr($data, $offset, 1));
        ++$offset;

        $attr = self::unpack(self::BIT_B8, substr($data, $offset, 1));
        ++$offset;

        $timestamp  = 0;
        $backOffset = $offset;

        try { // try unpack message format v1, falling back to v0 if it fails
            if ($magic >= self::MESSAGE_MAGIC_VERSION1) {
                $timestamp = self::unpack(self::BIT_B64, substr($data, $offset, 8));
                $offset   += 8;
            }

            $keyRet  = $this->decodeString(substr($data, $offset), self::BIT_B32);
            $offset += $keyRet['length'];

            $valueRet = $this->decodeString((string) substr($data, $offset), self::BIT_B32, $attr & Produce::COMPRESSION_CODEC_MASK);
            $offset  += $valueRet['length'];

            if ($offset !== $messageSize) {
                throw new Exception(
                    'pack message fail, message len:' . $messageSize . ' , data unpack offset :' . $offset
                );
            }
        } catch (Exception $e) { // try unpack message format v0
            $offset    = $backOffset;
            $timestamp = 0;
            $keyRet    = $this->decodeString(substr($data, $offset), self::BIT_B32);
            $offset   += $keyRet['length'];

            $valueRet = $this->decodeString(substr($data, $offset), self::BIT_B32);
            $offset  += $valueRet['length'];
        }

        return [
            'length' => $offset,
            'data'   => [
                'crc'       => $crc,
                'magic'     => $magic,
                'attr'      => $attr,
                'timestamp' => $timestamp,
                'key'       => $keyRet['data'],
                'value'     => $valueRet['data'],
            ],
        ];
    }
decode message
N.B., MessageSets are not preceded by an int32 like other array elements
in the protocol.

@return mixed[]|null