apache / apache/pulsar-client-python

Message corruption with `deliver_after`

未关闭
#41 9 条评论 0 个 reaction 已指派 1 人 已被 @BewareMyPower 认领 在 GitHub 查看
主要语言
Python
星标
75
派生
53
PR 合并指标
30 天内没有已合并 PR

描述

**Describe the bug**
Sending a message with compression, batching enabled and `deliver_after` results in a corrupted message. Disabling compression resolves the problem.

**To Reproduce**
Steps to reproduce the behavior:

Create a Pulsar Consumer subscribed to some topic
Start the consumer

Create a Pulsar Producer on the same topic with LZ4 compression enabled (see `Additional context`)
Call the `producer.send_async` method with `deliver_after` set to a valid timedelta (see `Additional context`)

The Consumer will now log something similar to:
```ERROR [140072584582912] ConsumerImpl:534 | [persistent://public/default/test, test, 0] Failed to decompress message with 6 at 96981:3```
```ERROR [140072584582912] ConsumerImpl:546 | [persistent://public/default/test, test, 0] Discarding corrupted message at 96981:3```

**Expected behavior**
Instead of throwing an error, the message should reach the consumer uncorrupted.

**Screenshots**

**Desktop (please complete the following information):**
- OS: Ubuntu 20.04.2 LTS
- CPU: Intel i7 8750h
- RAM: 16 GB DDR4
- Disk: 1 TB NVMe SSD (with over 100 GB free on the partition)
- Nvidia dGPU (20th gen)

**Additional context**
Using the Python Pulsar client version 2.7.0 (Python 3.8) and running Pulsar version 2.6.0 via Docker

Consumer options:
```
{
"consumer_type": pulsar.ConsumerType.Shared,
"initial_position": pulsar.InitialPosition.Earliest,
}
```

Producer options:
```
{
"compression_type": pulsar.CompressionType.LZ4,
"batching_enabled": True,
"block_if_queue_full": True,
"send_timeout_millis": 0
}
```

Producer call:
```
producer.send_async(
content=b'test',
callback=lambda res, msg_id: print(res),
partition_key=None,
deliver_after=datetime.timedelta(seconds=5.0)
)
```

贡献指南

打开贡献指南

评估

这个 Issue 还没有评估数据。

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。