apache / apache/pulsar-client-python

[BUG] use celery task send data to pulsar, init pulsar client error

Abierto
#150 3 comentarios 0 reacciones 0 asignados Ver en GitHub
Lenguaje dominante
Python
Estrellas
75
Forks
53
Métricas de merge de PR
Sin PR fusionados en 30 d

Descripción

python: 3.6
pulsar-client: pulsar-client[avro]==2.10.2
celery: 5.1.2

code example:
```python
#!/usr/bin/env python

import time
import random
import string
from pulsar import Client, CompressionType
from pulsar.schema import AvroSchema, Record, String, Integer

def generate_random_string(length=6):
charset = string.ascii_letters + string.digits
random_chars = random.choices(charset, k=length)
random_string = "".join(random_chars)
return random_string.capitalize()

class User(Record):
name = String()
age = Integer

UserAvroSchema = AvroSchema(User) # type: ignore

def gen_random_data():
return User(user=generate_random_string(), age=random.randint(0, 100))

class PulsarDemo(object):
def __init__(self) -> None:
self.SERVICE_URL = "pulsar://***"
self.TOPIC = "persistent://****"
client = Client(service_url=self.SERVICE_URL)
self.producer = client.create_producer(
topic=self.TOPIC,
schema=UserAvroSchema,
batching_enabled=True,
batching_max_messages=1000,
batching_max_publish_delay_ms=1000,
compression_type=CompressionType.SNAPPY, # type: ignore
)

def send_callback(self, send_result, msg_id):
print("Message published: result:{} msg_id:{}".format(send_result, msg_id))

def async_producer(self, cnt=1000):
while cnt >= 0:
data = gen_random_data()
self.producer.send_async(
data,
callback=self.send_callback,
)
time.sleep(0.01)
cnt -= 1
self.producer.flush()

# celery task
from celery import shared_task
@shared_task
def mock_data2pulsar(cnt=1000):
mock = PulsarDemo()
mock.async_producer()

```

an exception occurred at:
```text
[2023-08-29 10:27:46,933: ERROR/ForkPoolWorker-31] Pulsar error: TopicNotFound
```

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.