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

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

        if (! isset($payloads['max_wait_time'])) {
            $payloads['max_wait_time'] = 100; // default timeout 100ms
        }

        if (! isset($payloads['min_bytes'])) {
            $payloads['min_bytes'] = 64 * 1024; // 64k
        }

        $header = $this->requestHeader('kafka-php', self::FETCH_REQUEST, self::FETCH_REQUEST);
        $data   = self::pack(self::BIT_B32, (string) $payloads['replica_id']);
        $data  .= self::pack(self::BIT_B32, (string) $payloads['max_wait_time']);
        $data  .= self::pack(self::BIT_B32, (string) $payloads['min_bytes']);
        $data  .= self::encodeArray($payloads['data'], [$this, 'encodeFetchTopic']);
        $data   = self::encodeString($header . $data, self::PACK_INT32);

        return $data;
    }