不能获取到Header

[tw2066]

• 你遇到了什么问题？

不能获取到Header

• Kafka 环境是自建还是云服务？

自建

通过工具查询,头部已经设置上了

• 请执行下面的命令获取环境信息。

php -v & php --ri swoole & composer info | grep longlang/phpkafka

[1] 1067
[2] 1068
PHP 8.0.6 (cli) (built: Jun 21 2021 15:15:34) ( NTS )
Copyright (c) The PHP Group
Zend Engine v4.0.6, Copyright (c) Zend Technologies
    with Yasd v0.3.8, Our Copyright, by codinghuang

swoole

Swoole => enabled
Author => Swoole Team <team@swoole.com>
Version => 4.7.0
Built => Aug 12 2021 12:58:54
coroutine => enabled with boost asm context
epoll => enabled
eventfd => enabled
signalfd => enabled
cpu_affinity => enabled
spinlock => enabled
rwlock => enabled
sockets => enabled
openssl => OpenSSL 1.1.1f  31 Mar 2020
dtls => enabled
http2 => enabled
json => enabled
curl-native => enabled
pcre => enabled
zlib => 1.2.11
mutex_timedlock => enabled
pthread_barrier => enabled
futex => enabled
mysqlnd => enabled
async_redis => enabled

Directive => Local Value => Master Value
swoole.enable_coroutine => On => On
swoole.enable_library => On => On
swoole.enable_preemptive_scheduler => Off => Off
swoole.display_errors => On => On
swoole.use_shortname => Off => Off
swoole.unixsock_buffer_size => 8388608 => 8388608
longlang/phpkafka                       v1.1.5     A kafka client. Support php-fpm and Swoole.
[1]-  Done                    php -v
[2]+  Done                    php --ri swoole

• 提供最小可复现代码：

    public function send(Producer $producer)
    {

        $recordHeader = new RecordHeader();
        $recordHeader->setHeaderKey('test-h');
        $recordHeader->setValue('goods-info');
        $headers = [$recordHeader];
        $producer->send('local_hyperf', 'value', 'key',$headers);
        dump('send...');

    }

/**
 * @Consumer(topic="local_hyperf", nums=1, groupId="hyperf", autoCommit=true)
 */
class DemoKafkaConsumer extends AbstractConsumer
{
    public function consume(ConsumeMessage $message): string
    {
        dump($message->getHeaders());
        dump($message->getValue());
        Log::info(milliseconds() . ' ' . $message->getTopic() . ':' . $message->getKey() . ':' . $message->getValue());
        return Result::ACK;
    }
}

打印$message->getHeaders() 未空

[Yurunsoft]

推消息时候，$headers不用RecordHeader，直接数组就行：

$producer->send('local_hyperf', 'value', 'key', ['test-h' => 'goods-info']);

[tw2066]

数组会报错

[Yurunsoft]

确实有问题，晚点看看

[Yurunsoft]

2d497bf
已修复，longlang/phpkafka更新到dev-master试一下

[tw2066]

可以拿到了

array:1 [
  0 => longlang\phpkafka\Protocol\RecordBatch\RecordHeader^ {#4393
    #headerKey: "test-h"
    #value: "goods-info"
  }
]

[tw2066]

    $recordHeader = new RecordHeader();
    $recordHeader->setHeaderKey('test-h');
    $recordHeader->setValue('goods-info');
    $headers = [$recordHeader];
    $producer->send('local_hyperf', 'value', 'key',$headers);

可以考虑 优化一下 用数组

[Yurunsoft]

嗯，数组、RecordHeader都可以做一下兼容

[Yurunsoft]

已实现