apache / apache/pulsar-client-python

Message corruption with `deliver_after`

Abierto
#41 9 comentarios 0 reacciones 1 asignado Reclamado por @BewareMyPower Ver en GitHub
Lenguaje dominante
Python
Estrellas
75
Forks
53
Métricas de merge de PR
Sin PR fusionados en 30 d

Descripción

**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)
)
```

Guía de contribución

Abrir la guía de contribución

Evaluación

Este issue todavía no se ha evaluado.

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.